Skip to content
Snippets Groups Projects
Select Git revision
  • main default protected
1 result
Blame Permalink
test_resource.py 7.73 KiB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
101
102
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203 






import random
from uuid import uuid4

import pytest
from clowmdb.models import Resource, ResourceVersion
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload

from app.crud import CRUDResource
from app.schemas.resource import ResourceIn
from app.tests.utils.user import UserWithAuthHeader
from app.tests.utils.utils import CleanupList, random_lower_string


class TestResourceCRUDGet:
    @pytest.mark.asyncio
    async def test_get_resource(self, db: AsyncSession, random_resource: Resource) -> None:
        """
        Test for getting an existing resource from the database

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : clowmdb.models.Resource
            Random resource for testing.
        """
        resource = await CRUDResource.get(db, resource_id=random_resource.resource_id)
        assert resource is not None
        assert resource == random_resource

    @pytest.mark.asyncio
    async def test_get_non_existing_resource(self, db: AsyncSession) -> None:
        """
        Test for getting a non-existing resource from the database

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        """
        resource = await CRUDResource.get(db, resource_id=uuid4())
        assert resource is None


class TestResourceCRUDList:
    @pytest.mark.asyncio
    async def test_get_all_resources(self, db: AsyncSession, random_resource: Resource) -> None:
        """
        Test get all resources from the CRUD Repository.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : app.schemas.resource.ResourceOut
            Random resource for testing.
        """
        resources = await CRUDResource.list_resources(db)
        assert len(resources) == 1
        assert resources[0].resource_id == random_resource.resource_id

    @pytest.mark.asyncio
    async def test_get_resources_by_maintainer(
        self, db: AsyncSession, random_resource: Resource, random_user: UserWithAuthHeader
    ) -> None:
        """
        Test get only resources from the CRUD Repository by specific maintainer.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : app.schemas.resource.ResourceOut
            Random resource for testing.
        random_user : app.tests.utils.user.UserWithAuthHeader
            Random user for testing.
        """
        resources = await CRUDResource.list_resources(db, maintainer_id=random_user.user.uid)
        assert len(resources) == 1
        assert resources[0].resource_id == random_resource.resource_id

    @pytest.mark.asyncio
    async def test_get_resources_with_unpublished_version(self, db: AsyncSession, random_resource: Resource) -> None:
        """
        Test get only resources from the CRUD Repository with an unpublished version.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : app.schemas.resource.ResourceOut
            Random resource for testing.
        """
        resources = await CRUDResource.list_resources(db, version_status=[ResourceVersion.Status.LATEST])
        assert len(resources) == 1
        assert resources[0].resource_id == random_resource.resource_id

    @pytest.mark.asyncio
    async def test_get_resource_with_name_substring(self, db: AsyncSession, random_resource: Resource) -> None:
        """
        Test get resources with a substring in their name from the CRUD Repository.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : app.schemas.resource.ResourceOut
            Random resource for testing.
        """
        substring_indices = [0, 0]
        while substring_indices[0] == substring_indices[1]:
            substring_indices = sorted(random.choices(range(len(random_resource.name)), k=2))

        random_substring = random_resource.name[substring_indices[0] : substring_indices[1]]
        resources = await CRUDResource.list_resources(db, name_substring=random_substring)
        assert len(resources) > 0
        assert random_resource.resource_id in map(lambda w: w.resource_id, resources)

    @pytest.mark.asyncio
    async def test_search_non_existing_resource_by_name(self, db: AsyncSession, random_resource: Resource) -> None:
        """
        Test for getting a non-existing resource by its name from CRUD Repository.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : app.schemas.resource.ResourceOut
            Random resource for testing.
        """
        resources = await CRUDResource.list_resources(db, name_substring=2 * random_resource.name)
        assert sum(1 for w in resources if w.resource_id == random_resource.resource_id) == 0


class TestResourceCRUDDelete:
    @pytest.mark.asyncio
    async def test_delete_resource(self, db: AsyncSession, random_resource: Resource) -> None:
        """
        Test for deleting a resource from CRUD Repository.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_resource : app.schemas.resource.ResourceOut
            Random resource for testing.
        """
        await CRUDResource.delete(db, resource_id=random_resource.resource_id)

        deleted_resource = await db.scalar(
            select(Resource).where(Resource._resource_id == random_resource.resource_id.bytes)
        )
        assert deleted_resource is None

        deleted_resource_versions = (
            await db.scalars(
                select(ResourceVersion).where(ResourceVersion._resource_id == random_resource.resource_id.bytes)
            )
        ).all()
        assert len(deleted_resource_versions) == 0


class TestResourceCRUDCreate:
    @pytest.mark.asyncio
    async def test_create_resource(
        self, db: AsyncSession, random_user: UserWithAuthHeader, cleanup: CleanupList
    ) -> None:
        """
        Test for creating a new resource with the CRUD Repository.

        Parameters
        ----------
        db : sqlalchemy.ext.asyncio.AsyncSession.
            Async database session to perform query on.
        random_user : app.tests.utils.user.UserWithAuthHeader
            Random user for testing.
        cleanup : list[sqlalchemy.sql.dml.Delete]
            List were to append sql deletes that gets executed after the test
        """
        resource_in = ResourceIn(
            name=random_lower_string(8),
            description=random_lower_string(),
            source=random_lower_string(),
            release=random_lower_string(8),
        )

        resource = await CRUDResource.create(db, resource_in=resource_in, maintainer_id=random_user.user.uid)
        assert resource is not None
        cleanup.append(delete(Resource).where(Resource._resource_id == resource.resource_id.bytes))

        created_resource = await db.scalar(
            select(Resource)
            .where(Resource._resource_id == resource.resource_id.bytes)
            .options(joinedload(Resource.versions))
        )
        assert created_resource is not None
        assert created_resource == resource

        assert len(created_resource.versions) == 1
        assert created_resource.versions[0].release == resource_in.release