Coverage for tests/conftest.py: 51%

1import uuid 

2from asyncio import get_running_loop, new_event_loop 

3from unittest import mock 

4from unittest.mock import PropertyMock, patch 

5from uuid import UUID 

6 

7import bson 

8import pytest 

9import pytest_asyncio 

10import pytest_cases 

11from bson import UUID_SUBTYPE, Binary, UuidRepresentation 

12from dbcc import MongoTableEngine 

13from mongodb import clients_col, orgs_col, subsidiaries_col, users_col 

14from sotrans_models.models.admin.ads import AdvertisingBlockDBModel 

15from sotrans_models.models.admin.settings import AdminSettings 

16from sotrans_models.models.misc.client import ClientDBModel 

17from sotrans_models.models.misc.document import DocumentDBModel 

18from sotrans_models.models.organizations import ( 

19 CreateSubsidiaryModel, 

20 OrganizationDBModel, 

21 SubsidiaryDBModel, 

22) 

23from sotrans_models.models.resources.drivers import DriverDBModel 

24from sotrans_models.models.resources.trailers import TrailerDBModel 

25from sotrans_models.models.resources.trucks import TruckDBModel 

26from sotrans_models.models.roles import SotransRole 

27from sotrans_models.models.users import ( 

28 SotransOIDCUserModel, 

29 SotransUserDBModel, 

30) 

31from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams 

32 

33from .random_model import generate_models_json, generate_random_model_dict 

34 

35 

36class APIDataStorage: 

37 org: OrganizationDBModel | None = None 

38 driver: DriverDBModel | None = None 

39 truck: TruckDBModel | None = None 

40 trailer: TrailerDBModel | None = None 

41 ads: AdvertisingBlockDBModel | None = None 

42 admins_settings: AdminSettings | None = None 

43 document: DocumentDBModel | None = None 

44 subsidiary: SubsidiaryDBModel | None = None 

45 client: ClientDBModel | None = None 

46 

47 

48@pytest.fixture(scope="function") 

49def event_loop(): 

50 """Default for all async tests.""" 

51 try: 

52 loop = get_running_loop() 

53 except RuntimeError: 

54 loop = new_event_loop() 

55 yield loop 

56 loop.close() 

57 

58 

59@pytest_asyncio.fixture(scope="function") 

60async def mongo_table_engine(): 

61 mongo_db_table_engine = MongoTableEngine( 

62 "mongodb://localhost:27017", 

63 "sotrans_test", 

64 "test", 

65 mock_db_in_mem=True, 

66 ) 

67 yield mongo_db_table_engine 

68 await mongo_db_table_engine.db.client.drop_database("sotrans_test") 

69 

70 

71def side_effect_mongo_mock_from_uuid( 

72 uuid: UUID, uuid_representation=UuidRepresentation.STANDARD 

73): 

74 """Override (Mock) the bson.binary.Binary.from_uuid function to work for mongomock 

75 

76 Code is copy pasted from the original function, 

77 but `uuid_representation` is ignored and only code parts as if 

78 uuid_representation == UuidRepresentation.STANDARD are used 

79 """ 

80 if not isinstance(uuid, UUID): 

81 raise TypeError("uuid must be an instance of uuid.UUID") 

82 

83 subtype = UUID_SUBTYPE 

84 payload = uuid.bytes 

85 

86 return Binary(payload, subtype) 

87 

88 

89@pytest.fixture(autouse=True) 

90def mongo_uuid(monkeypatch): 

91 with mock.patch.object( 

92 Binary, "from_uuid", side_effect=side_effect_mongo_mock_from_uuid 

93 ): 

94 yield 

95 

96 

97@pytest_cases.fixture 

98def mock_get_list_params(): 

99 kwargs = { 

100 "where": None, 

101 "sort": None, 

102 "projection": None, 

103 "skip": 0, 

104 "limit": 20, 

105 "search": None, 

106 } 

107 return BaseGetListQueryParams(**kwargs) 

108 

109 

110@pytest.fixture 

111def mock_get_one_params(): 

112 kwargs = {"projection": None} 

113 return BaseGetOneQueryParams(**kwargs) 

114 

115 

116@pytest.fixture 

117def sotrans_user_roles(request): 

118 user_params = request.param 

119 with patch.object( 

120 SotransOIDCUserModel, "roles", return_value=user_params["roles"] 

121 ): 

122 random_data = generate_random_model_dict(SotransOIDCUserModel) 

123 print(random_data) 

124 print("9" * 999) 

125 random_data.update(user_params) 

126 print(random_data) 

127 sotrans_user_model = SotransOIDCUserModel(**random_data) 

128 yield sotrans_user_model 

129 

130 

131@pytest.fixture(scope="package") 

132def shared_uuid(): 

133 return uuid.uuid4() 

134 

135 

136@pytest.fixture(scope="package") 

137def shared_oid(): 

138 return bson.ObjectId() 

139 

140 

141@pytest.fixture(scope="package") 

142def shared_org_id(): 

143 return bson.ObjectId() 

144 

145 

146@pytest.fixture 

147def sotrans_company_dir(shared_uuid): 

148 with patch.object( 

149 SotransOIDCUserModel, "roles", new_callable=PropertyMock 

150 ) as mock_roles: 

151 mock_roles.return_value = [SotransRole.company_director] 

152 random_data = generate_random_model_dict(SotransOIDCUserModel) 

153 random_data.update(sub=shared_uuid) 

154 sotrans_user_model = SotransOIDCUserModel(**random_data) 

155 yield sotrans_user_model 

156 

157 

158@pytest.fixture 

159def sotrans_company_logisitician(shared_uuid, shared_org_id): 

160 with patch.object( 

161 SotransOIDCUserModel, "roles", new_callable=PropertyMock 

162 ) as mock_roles: 

163 mock_roles.return_value = [SotransRole.company_logistician] 

164 random_data = generate_random_model_dict(SotransOIDCUserModel) 

165 random_data.update(sub=shared_uuid, organization_id=str(shared_org_id)) 

166 sotrans_user_model = SotransOIDCUserModel(**random_data) 

167 yield sotrans_user_model 

168 

169 

170@pytest_asyncio.fixture 

171async def tmp_user(): 

172 model_dump = generate_models_json( 

173 SotransUserDBModel, 

174 { 

175 "id": uuid.uuid4(), 

176 # "subsidiary_id": str(tmp_subsidiary.id) 

177 }, 

178 ) 

179 await users_col.collection.insert_one(model_dump) 

180 user = SotransUserDBModel(**model_dump) 

181 yield user 

182 await users_col.collection.delete_one({"id": user.id}) 

183 

184 

185@pytest_asyncio.fixture 

186async def tmp_org(tmp_user): 

187 model_dump = generate_models_json( 

188 OrganizationDBModel, specific={"employees": [tmp_user.model_dump()]} 

189 ) 

190 del model_dump["id"] 

191 created = await orgs_col.create(model_dump) 

192 org = OrganizationDBModel(**created) 

193 yield org 

194 await orgs_col.delete_by_id(org.id) 

195 

196 

197@pytest_asyncio.fixture 

198async def tmp_subsidiary(tmp_user): 

199 sub = generate_models_json( 

200 CreateSubsidiaryModel, {"employees": [tmp_user.model_dump()]} 

201 ) 

202 created = await subsidiaries_col.create(sub) 

203 sm = SubsidiaryDBModel(**created) 

204 yield sm 

205 await subsidiaries_col.delete_by_id(sm.id) 

206 

207 

208@pytest_asyncio.fixture 

209async def tmp_client(): 

210 cl = generate_models_json(ClientDBModel, {"inn": "1231231234"}) 

211 created = await clients_col.create(cl) 

212 clm = ClientDBModel(**created) 

213 yield clm 

214 await clients_col.delete_by_id(clm.id)