trolldb.test_utils.mongodb_database module

The module which provides testing utilities to make MongoDB databases/collections and fill them with test data.

trolldb.test_utils.mongodb_database.mongodb_for_test_context(database_config: DatabaseConfig = pydantic.BaseModel.database) → Generator[MongoClient, Any, None]

A context manager for the MongoDB client given test configurations.

Note

This is based on Pymongo and not the motor async driver. For testing purposes this is sufficient, and we do not need async capabilities.

Parameters:

database_config (Optional, default test_app_config.database) – The configuration object for the database.

Yields:

MongoClient – The MongoDB client object (from Pymongo)

class trolldb.test_utils.mongodb_database.Time

Bases: object

A static class to enclose functionalities for generating random timestamps.

min_start_time: ClassVar[datetime] = datetime.datetime(2019, 1, 1, 0, 0)

The minimum timestamp which is allowed to appear in our data.

max_end_time: ClassVar[datetime] = datetime.datetime(2024, 1, 1, 0, 0)

The maximum timestamp which is allowed to appear in our data.

delta_time: ClassVar[int] = 157766400

The difference between the maximum and minimum timestamps in seconds.

static random_interval_secs(max_interval_secs: int) → timedelta

Generates a random time interval between zero and the given max interval in seconds.

static random_start_time() → datetime

Generates a random start time.

The start time has a lower bound which is specified by min_start_time and an upper bound given by max_end_time.

static random_end_time(start_time: datetime, max_interval_secs: int = 300) → datetime

Generates a random end time.

The end time is within max_interval_secs seconds from the given start_time. By default, the interval is set to 300 seconds (5 minutes).

class trolldb.test_utils.mongodb_database.Document(platform_name: str, sensor: str)

Bases: object

A class which defines functionalities to generate database documents/data which are similar to real data.

__init__(platform_name: str, sensor: str) → None

Initializes the document given its platform and sensor names.

generate_dataset(max_count: int) → list[dict]

Generates the dataset for a given document.

This corresponds to the list of files which are stored in each document. The number of items in a dataset is randomly chosen from 1 to max_count for each document.

like_mongodb_document() → dict

Returns a dictionary which resembles the format we have for our real data when saving them to MongoDB.

class trolldb.test_utils.mongodb_database.TestDatabase

Bases: object

A static class which encloses functionalities to prepare and fill the test database with test data.

unique_platform_names: ClassVar[list[str]] = ['PA', 'PB', 'PC']

The unique platform names that will be used to generate the sample of all platform names.

platform_names: ClassVar[list[str]] = ['PA', 'PC', 'PB', 'PB', 'PC', 'PC', 'PA', 'PC', 'PB', 'PC', 'PC', 'PA', 'PA', 'PA', 'PB', 'PB', 'PA', 'PA', 'PA', 'PB']

Example platform names.

Warning

The value of this variable changes randomly every time. What you see above is just an example which has been generated as a result of building the documentation!

unique_sensors: ClassVar[list[str]] = ['SA', 'SB', 'SC']

The unique sensor names that will be used to generate the sample of all sensor names.

sensors: ClassVar[list[str]] = ['SC', 'SC', 'SA', 'SC', 'SC', 'SC', 'SA', 'SA', 'SA', 'SA', 'SC', 'SB', 'SC', 'SB', 'SB', 'SA', 'SB', 'SB', 'SA', 'SB']

Example sensor names.

Warning

The value of this variable changes randomly every time. What you see above is just an example which has been generated as a result of building the documentation!

database_names: ClassVar[list[str]] = [pydantic.BaseModel.database.main_database_name, 'another_test_database']

List of all database names.

The first element is the main database that will be queried by the API and includes the test data. The second database is for testing scenarios when one attempts to access another existing database or collection.

collection_names: ClassVar[list[str]] = [pydantic.BaseModel.database.main_collection_name, 'another_test_collection']

List of all collection names.

The first element is the main collection that will be queried by the API and includes the test data. The second collection is for testing scenarios when one attempts to access another existing collection.

all_database_names: ClassVar[list[str]] = ['admin', 'config', 'local', pydantic.BaseModel.database.main_database_name, 'another_test_database']

All database names including the default ones which are automatically created by MongoDB.

documents: ClassVar[list[dict]] = []

The list of documents which include test data.

classmethod generate_documents(random_shuffle: bool = True) → None

Generates test documents which for practical purposes resemble real data.

Warning

This method is not pure! The side effect is that the TestDatabase.documents is reset to new values.

classmethod reset() → None

Resets all the databases/collections.

This is done by deleting all documents in the collections and then inserting a single empty document, i.e. {}, in them.

classmethod write_test_data() → None

Fills databases/collections with test data.

classmethod get_documents_from_database() → list[dict]

Retrieves all the documents from the database.

Returns:

A list of all documents from the database. This matches the content of documents with the addition of IDs which are assigned by the MongoDB.

classmethod get_document_ids_from_database() → list[str]

Retrieves all the document IDs from the database.

classmethod find_min_max_datetime() → dict[str, dict]

Finds the minimum and the maximum for both the start_time and the end_time.

We use brute force for this purpose. We set the minimum to a large value (year 2100) and the maximum to a small value (year 1900). We then iterate through all documents and update the extrema.

Returns:

A dictionary whose schema matches the response returned by the /datetime route of the API.

classmethod _query_platform_sensor(document, platform=None, sensor=None) → bool

An auxiliary method to the TestDatabase.match_query().

classmethod _query_time(document, time_min=None, time_max=None) → bool

An auxiliary method to the TestDatabase.match_query().

classmethod match_query(platform=None, sensor=None, time_min=None, time_max=None) → list[str]

Matches the given query.

We first take all the documents and then progressively remove all that do not match the given queries until we end up with those that match. When a query is None, it does not have any effect on the results. This method will be used in testing the /queries route of the API.

classmethod prepare() → None

Prepares the MongoDB instance by first resetting the database and filling it with generated test data.