M2-4964: Add option to save users session fixtures after tests

src/apps/job/tests/test_jobs_services.py

@@ -5,7 +5,7 @@
 from apps.job.crud import JobCRUD
 from apps.job.db.schemas import JobSchema
 from apps.job.service import JobService
-from apps.users.db.schemas import UserSchema
+from apps.users.domain import User
 
 
 @pytest.mark.parametrize("status,", (JobStatus.in_progress, JobStatus.pending, JobStatus.retry))
@@ -54,31 +54,29 @@ async def test_change_status_without_details(
     assert job_updated.status == JobStatus.success
 
 
-async def test_get_or_create_owned__create_job_default_status_is_pending(
-    session: AsyncSession, tom: UserSchema
-) -> None:
-    job_name = "tomjob"
-    srv = JobService(session, tom.id)
+async def test_get_or_create_owned__create_job_default_status_is_pending(session: AsyncSession, user: User) -> None:
+    job_name = "userjob"
+    srv = JobService(session, user.id)
     job = await srv.get_or_create_owned(job_name)
     assert job.status == JobStatus.pending
-    assert job.creator_id == tom.id
+    assert job.creator_id == user.id
     assert job.name == job_name
 
 
-async def test_get_or_create_owned__create_job_with_status_in_progress(session: AsyncSession, tom: UserSchema) -> None:
-    job_name = "tomjob"
-    srv = JobService(session, tom.id)
+async def test_get_or_create_owned__create_job_with_status_in_progress(session: AsyncSession, user: User) -> None:
+    job_name = "userjob"
+    srv = JobService(session, user.id)
     job = await srv.get_or_create_owned(job_name, status=JobStatus.in_progress)
     assert job.status == JobStatus.in_progress
 
 
 async def test_get_or_create_owned__create_job_with_details(
     session: AsyncSession,
-    tom: UserSchema,
+    user: User,
     job_details: dict[str, str],
 ) -> None:
-    job_name = "tomjob"
-    srv = JobService(session, tom.id)
+    job_name = "userjob"
+    srv = JobService(session, user.id)
     job = await srv.get_or_create_owned(job_name, details=job_details)
     assert job.details == job_details
 

src/apps/logs/tests.py

@@ -1,4 +1,3 @@
-import uuid
 
 from pytest import fixture, mark
 
@@ -7,23 +6,23 @@
 EMPTY_DESCRIPTIONS = [
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test1",
         notification_descriptions=None,
         notification_in_queue=[{"name": "in_queue1"}],
         scheduled_notifications=[{"name": "notifications1"}],
     ),
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test2",
         notification_descriptions=[],
         notification_in_queue=[{"name": "in_queue2"}],
         scheduled_notifications=[{"name": "notifications2"}],
     ),
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test3",
         notification_descriptions=None,
         notification_in_queue=[{"name": "in_queue2"}],
@@ -34,23 +33,23 @@
 EMPTY_QUEUE = [
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test1",
         notification_descriptions=[{"name": "description"}],
         notification_in_queue=None,
         scheduled_notifications=[{"name": "notifications1"}],
     ),
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test2",
         notification_descriptions=[{"name": "description"}],
         notification_in_queue=[],
         scheduled_notifications=[{"name": "notifications2"}],
     ),
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test3",
         notification_descriptions=[{"name": "description"}],
         notification_in_queue=None,
@@ -61,23 +60,23 @@
 EMPTY_SCHEDULE = [
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test1",
         notification_descriptions=[{"name": "description"}],
         notification_in_queue=[{"name": "in_queue1"}],
         scheduled_notifications=None,
     ),
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test2",
         notification_descriptions=[{"name": "description"}],
         notification_in_queue=[{"name": "in_queue2"}],
         scheduled_notifications=[],
     ),
     dict(
         user_id="[email protected]",
-        device_id=str(uuid.uuid4()),
+        device_id="deviceid",
         action_type="test3",
         notification_descriptions=[{"name": "description"}],
         notification_in_queue=[{"name": "in_queue2"}],
@@ -91,7 +90,7 @@ def dummy_logs_payload() -> list[dict]:
     return [
         dict(
             user_id="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id="deviceid",
             action_type=f"test{i}",
             notification_descriptions=[{"sample": f"descriptions{i}"}],
             notification_in_queue=[{"sample": f"queue{i}"}],
@@ -103,14 +102,11 @@ def dummy_logs_payload() -> list[dict]:
 
 class TestNotificationLogs(BaseTest):
     logs_url = "/logs/notification"
-    fixtures = [
-        "users/fixtures/user_devices.json",
-    ]
 
-    async def test_create_log(self, client):
+    async def test_create_log(self, client, device_tom):
         create_data = dict(
             user_id="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id="deviceid",
             action_type="test",
             notification_descriptions=[{"sample": "json"}],
             notification_in_queue=[{"sample": "json"}],
@@ -121,20 +117,21 @@ async def test_create_log(self, client):
         assert response.status_code == 201, response.json()
         assert response.json()["result"]["id"]
 
-    async def test_retrieve_log(self, client):
+    async def test_retrieve_log(self, client, device_tom):
         query = dict(
             email="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id=device_tom,
         )
 
         response = await client.get(self.logs_url, query=query)
 
         assert response.status_code == 200, response.json()
         assert isinstance(response.json()["result"], list)
 
+        new_device_id = "new_device_id"
         create_data = dict(
             user_id="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id=new_device_id,
             action_type="test",
             notification_descriptions=[{"sample": "json"}],
             notification_in_queue=[{"sample": "json"}],
@@ -147,7 +144,7 @@ async def test_retrieve_log(self, client):
 
         query = dict(
             email="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id=new_device_id,
             limit=10,
         )
 
@@ -178,7 +175,7 @@ async def test_create_log_use_previous_value_if_attribute_null(
 
         create_data = dict(
             user_id="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id="deviceid",
             action_type="test",
             notification_descriptions=description,
             notification_in_queue=queue,
@@ -198,7 +195,7 @@ async def test_create_log_use_none_value_if_attribute_null_at_first_log(self, cl
             self.logs_url,
             data=dict(
                 user_id="[email protected]",
-                device_id=str(uuid.uuid4()),
+                device_id="deviceid",
                 action_type="test",
                 notification_descriptions=None,
                 notification_in_queue=[{"name": "notification_in_queue"}],
@@ -216,15 +213,15 @@ async def test_create_log_use_previous_non_null_if_attribute_null(self, client):
         payloads = [
             dict(
                 user_id="[email protected]",
-                device_id=str(uuid.uuid4()),
+                device_id="deviceid",
                 action_type="test",
                 notification_descriptions=[{"name": "descriptions1"}],
                 notification_in_queue=[{"name": "in_queue1"}],
                 scheduled_notifications=[{"name": "notifications1"}],
             ),
             dict(
                 user_id="[email protected]",
-                device_id=str(uuid.uuid4()),
+                device_id="deviceid",
                 action_type="test",
                 notification_descriptions=None,
                 notification_in_queue=[{"name": "in_queue2"}],
@@ -237,7 +234,7 @@ async def test_create_log_use_previous_non_null_if_attribute_null(self, client):
 
         create_data = dict(
             user_id="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id="deviceid",
             action_type="test",
             notification_descriptions=None,
             notification_in_queue=[{"name": "in_queue3"}],
@@ -256,15 +253,15 @@ async def test_create_log_allow_empty_array(self, client):
         payloads = [
             dict(
                 user_id="[email protected]",
-                device_id=str(uuid.uuid4()),
+                device_id="deviceid",
                 action_type="test",
                 notification_descriptions=[{"name": "descriptions1"}],
                 notification_in_queue=[{"name": "in_queue1"}],
                 scheduled_notifications=[{"name": "notifications1"}],
             ),
             dict(
                 user_id="[email protected]",
-                device_id=str(uuid.uuid4()),
+                device_id="deviceid",
                 action_type="test",
                 notification_descriptions=[],
                 notification_in_queue=[{"name": "in_queue2"}],
@@ -278,7 +275,7 @@ async def test_create_log_allow_empty_array(self, client):
 
         query = dict(
             email="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id="deviceid",
             limit=5,
         )
 
@@ -309,7 +306,7 @@ async def test_create_log_allow_empty_array_if_prev_is_none(self, client, param,
 
         query = dict(
             email="[email protected]",
-            device_id=str(uuid.uuid4()),
+            device_id="deviceid",
             limit=5,
         )
 

src/apps/users/services/user.py

@@ -15,13 +15,13 @@ def __init__(self, session) -> None:
         self.session = session
 
     # TODO: Remove later, keep for now for backward compatibility for tests
-    async def create_superuser(self, uuid_: uuid.UUID = uuid.uuid4()) -> None:
+    async def create_superuser(self, test_id: uuid.UUID = uuid.uuid4()) -> None:
         crud = UsersCRUD(self.session)
         super_admin = await crud.get_super_admin()
         # Let's keep this frozen feature
         if super_admin is None:
             super_admin = UserSchema(
-                id=uuid_,
+                id=test_id,
                 email=hash_sha224(settings.super_admin.email),
                 first_name=settings.super_admin.first_name,
                 last_name=settings.super_admin.last_name,
@@ -38,12 +38,12 @@ async def create_superuser(self, uuid_: uuid.UUID = uuid.uuid4()) -> None:
             await UserWorkspaceCRUD(self.session).save(schema=workspace)
 
     # TODO: Remove later, keep for now for backward compatibility for tests
-    async def create_anonymous_respondent(self, uuid_: uuid.UUID = uuid.uuid4()) -> None:
+    async def create_anonymous_respondent(self, test_id: uuid.UUID = uuid.uuid4()) -> None:
         crud = UsersCRUD(self.session)
         anonymous_respondent = await crud.get_anonymous_respondent()
         if not anonymous_respondent:
             anonymous_respondent = UserSchema(
-                id=uuid_,
+                id=test_id,
                 email=hash_sha224(settings.anonymous_respondent.email),
                 first_name=settings.anonymous_respondent.first_name,
                 last_name=settings.anonymous_respondent.last_name,
@@ -55,16 +55,26 @@ async def create_anonymous_respondent(self, uuid_: uuid.UUID = uuid.uuid4()) ->
             )
             await crud.save(anonymous_respondent)
 
-    async def create_user(self, data: UserCreate) -> User:
-        user_schema = await UsersCRUD(self.session).save(
-            UserSchema(
+    # TODO: remove test_id, when all JSON fixtures are deleted
+    async def create_user(self, data: UserCreate, test_id: uuid.UUID | None = None) -> User:
+        if test_id is not None:
+            schema = UserSchema(
+                id=test_id,
                 email=data.hashed_email,
                 first_name=data.first_name,
                 last_name=data.last_name,
                 hashed_password=data.hashed_password,
                 email_encrypted=data.email,
             )
-        )
+        else:
+            schema = UserSchema(
+                email=data.hashed_email,
+                first_name=data.first_name,
+                last_name=data.last_name,
+                hashed_password=data.hashed_password,
+                email_encrypted=data.email,
+            )
+        user_schema = await UsersCRUD(self.session).save(schema)
 
         user: User = User.from_orm(user_schema)
         return user

src/apps/users/tests/fixtures/user_devices.py

@@ -0,0 +1,22 @@
+from typing import cast
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from apps.users.cruds.user_device import UserDevicesCRUD
+from apps.users.db.schemas import UserDeviceSchema
+from apps.users.domain import User
+from apps.users.services.user_device import UserDeviceService
+
+
+@pytest.fixture(scope="session", autouse=True)
+async def device_tom(tom: User, global_session: AsyncSession):
+    service = UserDeviceService(global_session, tom.id)
+    await service.add_device("deviceid")
+    await global_session.commit()
+    crud = UserDevicesCRUD(global_session)
+    device = await crud._get("device_id", "deviceid")
+    device = cast(UserDeviceSchema, device)
+    yield device.device_id
+    await service.remove_device("deviceid")
+    await global_session.commit()