Skip to content

fediverse_pasture.runner.result_store

ResultStore

Source code in fediverse_pasture/runner/result_store.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class ResultStore:
    async def add_result(self, test_name: str, application_name: str, data: dict):
        """Adds a result to the database. The pairs (test_name, application_name)
        are assumed to be unique. If data already exists, it is overwritten with
        the new data.

        :param test_name: Name of the test
        :param application_name: Name of the application
        :param data: Data to add
        """
        await TestRecord.update_or_create(
            test_name=test_name,
            application_name=application_name,
            defaults={"data": data},
        )

    async def delete_record(self, test_name: str, application_name: str):
        """Deletes database record if exists

        :param test_name: Name of the test
        :param application_name: Application to delete the result for
        """
        record = await TestRecord.get_or_none(
            test_name=test_name,
            application_name=application_name,
        )
        if record:
            await record.delete()

    async def results_for_test(self, test_name: str) -> list:
        """Retrieves the results for a given test_name"""
        result = await TestRecord.filter(test_name=test_name).all()

        return [{"application_name": x.application_name, **x.data} for x in result]

    async def entry_for_test(self, test_name: str) -> Entry:
        """Returns an Entry for test_name. We note here that an
        Entry contains the results for each application the test result
        is available for.

        :param test_name: name of the test entry
        :return: Entry from record
        """
        result = await TestRecord.filter(test_name=test_name).all()

        return Entry({x.application_name: x.data for x in result})

    async def as_list(self) -> list:
        records = await TestRecord.filter().order_by("test_name", "application_name")

        return [
            {
                "test_name": record.test_name,
                "application_name": record.application_name,
                "data": json.dumps(record.data, indent=2),
            }
            for record in records
        ]

    async def load(self, filename: str = "test_results.toml"):
        """Deletes the current content from the database, then loads
        the content from filename to it.

        :param filename: filename to load from"""

        await TestRecord.filter().delete()

        with open(filename, "rb") as fp:
            data = tomllib.load(fp)

        for x in data["entries"]:
            await self.add_result(
                test_name=x["test_name"],
                application_name=x["application_name"],
                data=json.loads(x["data"]),
            )

    async def save(self, filename: str = "test_results.toml"):
        """Saves the content of the database to the file given by filename

        :param filename: filename to save to
        """

        with open(filename, "wb") as fp:
            tomli_w.dump({"entries": await self.as_list()}, fp, multiline_strings=True)

add_result(test_name, application_name, data) async

Adds a result to the database. The pairs (test_name, application_name) are assumed to be unique. If data already exists, it is overwritten with the new data.

Parameters:

Name Type Description Default
test_name str

Name of the test

 required
application_name str

Name of the application

 required
data dict

Data to add

 required
Source code in fediverse_pasture/runner/result_store.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
async def add_result(self, test_name: str, application_name: str, data: dict):
    """Adds a result to the database. The pairs (test_name, application_name)
    are assumed to be unique. If data already exists, it is overwritten with
    the new data.

    :param test_name: Name of the test
    :param application_name: Name of the application
    :param data: Data to add
    """
    await TestRecord.update_or_create(
        test_name=test_name,
        application_name=application_name,
        defaults={"data": data},
    )

delete_record(test_name, application_name) async

Deletes database record if exists

Parameters:

Name Type Description Default
test_name str

Name of the test

 required
application_name str

Application to delete the result for

 required
Source code in fediverse_pasture/runner/result_store.py 
32
33
34
35
36
37
38
39
40
41
42
43
async def delete_record(self, test_name: str, application_name: str):
    """Deletes database record if exists

    :param test_name: Name of the test
    :param application_name: Application to delete the result for
    """
    record = await TestRecord.get_or_none(
        test_name=test_name,
        application_name=application_name,
    )
    if record:
        await record.delete()

entry_for_test(test_name) async

Returns an Entry for test_name. We note here that an Entry contains the results for each application the test result is available for.

Parameters:

Name Type Description Default
test_name str

name of the test entry

 required

Returns:

Type Description
Entry

Entry from record
Source code in fediverse_pasture/runner/result_store.py 
51
52
53
54
55
56
57
58
59
60
61
async def entry_for_test(self, test_name: str) -> Entry:
    """Returns an Entry for test_name. We note here that an
    Entry contains the results for each application the test result
    is available for.

    :param test_name: name of the test entry
    :return: Entry from record
    """
    result = await TestRecord.filter(test_name=test_name).all()

    return Entry({x.application_name: x.data for x in result})

load(filename='test_results.toml') async

Deletes the current content from the database, then loads the content from filename to it.

Parameters:

Name Type Description Default
filename str

filename to load from

 'test_results.toml'
Source code in fediverse_pasture/runner/result_store.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def load(self, filename: str = "test_results.toml"):
    """Deletes the current content from the database, then loads
    the content from filename to it.

    :param filename: filename to load from"""

    await TestRecord.filter().delete()

    with open(filename, "rb") as fp:
        data = tomllib.load(fp)

    for x in data["entries"]:
        await self.add_result(
            test_name=x["test_name"],
            application_name=x["application_name"],
            data=json.loads(x["data"]),
        )

results_for_test(test_name) async

Retrieves the results for a given test_name

Source code in fediverse_pasture/runner/result_store.py 
45
46
47
48
49
async def results_for_test(self, test_name: str) -> list:
    """Retrieves the results for a given test_name"""
    result = await TestRecord.filter(test_name=test_name).all()

    return [{"application_name": x.application_name, **x.data} for x in result]

save(filename='test_results.toml') async

Saves the content of the database to the file given by filename

Parameters:

Name Type Description Default
filename str

filename to save to

 'test_results.toml'
Source code in fediverse_pasture/runner/result_store.py 
 93
 94
 95
 96
 97
 98
 99
100
async def save(self, filename: str = "test_results.toml"):
    """Saves the content of the database to the file given by filename

    :param filename: filename to save to
    """

    with open(filename, "wb") as fp:
        tomli_w.dump({"entries": await self.as_list()}, fp, multiline_strings=True)

with_store(db_url='sqlite://test_results.sqlite') async

Initializes the database and returns a ResultStore. Usage:

async with with_store() as store:
    await store.add_result(...)
    ...
    await store.results_for_test(...)

Parameters:

Name Type Description Default
db_url str

Database url string

 'sqlite://test_results.sqlite'
Source code in fediverse_pasture/runner/result_store.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@asynccontextmanager
async def with_store(db_url: str = "sqlite://test_results.sqlite") -> ResultStore:
    """Initializes the database and returns a ResultStore. Usage:

    ```python
    async with with_store() as store:
        await store.add_result(...)
        ...
        await store.results_for_test(...)
    ```

    :param db_url: Database url string
    """
    await Tortoise.init(
        config={
            "connections": {"default": db_url},
            "apps": {
                "models": {
                    "models": [
                        "fediverse_pasture.runner.models",
                    ],
                    "default_connection": "default",
                },
            },
        },
    )
    await Tortoise.generate_schemas()

    yield ResultStore()

    await Tortoise.close_connections()