logos-messaging/logos-messaging-interop-tests
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-interop-tests.git synced 2026-01-10 01:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

postgress test

Browse Source
This commit is contained in:
Florin Barbu 2024-05-27 17:30:24 +03:00
parent 1df8cfb8f2
commit e8a165c71e
No known key found for this signature in database
GPG Key ID: 593D6DBC6D9E5095
2 changed files with 17 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@ -29,6 +29,7 @@ pytest-rerunfailures==13.0
pytest-timeout==2.2.0
pytest-xdist==3.5.0
python-dotenv==1.0.1
pytest-dependency==0.6.0
PyYAML==6.0.1
requests==2.31.0
setuptools==69.0.3

20
tests/store/test_external_db.py
View File

@ -6,13 +6,25 @@ from src.env_vars import PG_PASS, PG_USER
logger = get_custom_logger(__name__)
@pytest.mark.usefixtures("start_postgres_container")
class TestExternalDb(StepsStore):
def test_postgres_db(self):
self.setup_first_publishing_node(store="true", relay="true", store_message_db_url=f"postgres://{PG_USER}:{PG_PASS}@postgres:5432/postgres")
self.setup_first_store_node(store="true", relay="true")
postgress_url = f"postgres://{PG_USER}:{PG_PASS}@postgres:5432/postgres"
@pytest.fixture(scope="function", autouse=True)
def node_postgres_setup(self, store_setup, start_postgres_container):
self.setup_first_publishing_node(store="true", relay="true", store_message_db_url=self.postgress_url)
self.setup_first_store_node(store="false", relay="true")
self.subscribe_to_pubsub_topics_via_relay()
@pytest.mark.dependency(name="test_on_empty_postgress_db")
def test_on_empty_postgress_db(self):
message = self.create_message()
self.publish_message(message=message)
self.check_published_message_is_stored(page_size=5, ascending="true")
assert len(self.store_response.messages) == 1
@pytest.mark.dependency(depends=["test_on_empty_postgress_db"])
def test_on_postgress_db_with_one_message(self):
message = self.create_message()
self.publish_message(message=message)
self.check_published_message_is_stored(page_size=5, ascending="true")
assert len(self.store_response.messages) == 2