lavkach3

1
import asyncio
2
from datetime import datetime
3
from typing import Type, Dict, Any, List
4

5
import pytest
6
import pytest_asyncio
7
from httpx import AsyncClient
8
from sqlalchemy import text
9
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
10
from sqlalchemy.orm import sessionmaker
11
from starlette.requests import Request
12

13
from app.basic.company.models import Company
14
from app.basic.company.schemas import CompanyCreateScheme
15
from app.basic.company.services import CompanyService
16
from app.basic.product.models import ProductType
17
from app.basic.store.schemas import StoreCreateScheme
18
from app.basic.store.services import StoreService
19
from app.basic.product.services import ProductCategoryService, ProductStorageTypeService, ProductService
20
from app.basic.product.schemas import ProductCategoryCreateScheme, ProductStorageTypeCreateScheme, ProductCreateScheme
21
from app.basic.uom.models import UomType
22
from app.basic.uom.schemas import UomCategoryCreateScheme, UomCreateScheme
23
from app.basic.uom.services import UomCategoryService, UomService
24
from app.basic.user.models import User
25
from app.basic.user.schemas import UserCreateScheme, RoleCreateScheme, LoginResponseSchema
26
from app.basic.user.services import UserService
27
from app.basic.user.services.role_service import RoleService
28
from app.basic.basic_server import app
29
from app.inventory.inventory_server import app as basic_app
30
from app.inventory.location.enums import LocationClass, PutawayStrategy
31
from app.inventory.location.schemas import LocationTypeCreateScheme, LocationCreateScheme
32
from app.inventory.location.services import LocationService, LocationTypeService
33
from app.inventory.order.schemas import OrderTypeCreateScheme
34
from app.inventory.order.services import OrderTypeService
35
from app.inventory.quant.schemas import LotCreateScheme, QuantCreateScheme
36
from app.inventory.quant.services import LotService, QuantService
37
from core.config import config
38
from core.db.session import Base
39
from core.fastapi.schemas import CurrentUser
40
from core.permissions import permits
41
from core.service import base
42
from core.service.base import ModelType
43

44

45
@pytest.fixture(scope="session")
46
def event_loop() -> asyncio.AbstractEventLoop:
47
    """Create an instance of the default event loop for each test case."""
48
    loop = asyncio.get_event_loop_policy().new_event_loop()
49
    yield loop
50
    loop.close()
51

52

53
@pytest.fixture(scope="session")
54
def engine():
55
    engine = create_async_engine(config.TEST_WRITER_DB_URL)
56
    yield engine
57
    engine.sync_engine.dispose()
58

59

60
@pytest_asyncio.fixture(scope="session")
61
async def prepare_db():
62
    create_db_engine = create_async_engine(
63
        config.POSTGRES_TEST_DATABASE_URL,
64
        isolation_level="AUTOCOMMIT",
65
    )
66
    db_name_base = config.DB_NAME_TEST
67
    for i in range(5):
68
        db_name = f'{db_name_base}_{i}'
69
        try:
70
            async with create_db_engine.begin() as connection:
71
                await connection.execute(
72
                    text(
73
                        "drop database if exists {name};".format(
74
                            name=db_name
75
                        )
76
                    ),
77
                )
78
            await connection.execute(
79
                text("create database {name};".format(name=db_name)),
80
            )
81
        except Exception as ex:
82
            continue
83

84

85

86
@pytest_asyncio.fixture(scope="session")
87
async def db_session(engine) -> AsyncSession:
88
    async with engine.begin() as connection:
89
        await connection.run_sync(Base.metadata.drop_all)
90
        await connection.run_sync(Base.metadata.create_all)
91
        TestingSessionLocal = sessionmaker(
92
            expire_on_commit=False,
93
            class_=AsyncSession,
94
            bind=engine,
95
        )
96
        async with TestingSessionLocal(bind=connection) as session:
97
            yield session
98
            await session.flush()
99
            await session.rollback()
100

101

102
@pytest.fixture(scope="session")
103
def override_get_db(prepare_db, db_session: AsyncSession):
104
    async def _override_get_db():
105
        yield db_session
106

107
    return _override_get_db
108

109

110
@pytest_asyncio.fixture(scope="session")
111
def ovveride_base_init(db_session):
112
    def __init__(self, request: Request | CurrentUser | None, model: Type[ModelType], session=None):
113
        if isinstance(request, CurrentUser):
114
            self.user = request
115
        elif isinstance(request, Request):
116
            self.user = request.user
117
        self.model = model
118
        self.session = db_session
119

120
    return __init__
121

122

123
@pytest_asyncio.fixture(scope="session")
124
async def async_client(override_get_db, db_session, ovveride_base_init):
125
    base.BaseService.__init__ = ovveride_base_init
126
    async with AsyncClient(app=app, base_url="http://test") as ac:
127
        yield ac
128

129
@pytest_asyncio.fixture(scope="session")
130
async def async_inventory_client(override_get_db, db_session, ovveride_base_init):
131
    base.BaseService.__init__ = ovveride_base_init
132
    async with AsyncClient(app=basic_app, base_url="http://test") as ac:
133
        yield ac
134

135

136
@pytest_asyncio.fixture
137
async def user_admin(db_session: AsyncSession) -> User:
138
    user = UserCreateScheme(**{
139
        "email": "admin@admin.com",
140
        "nickname": "admin",
141
        "locale": "RU",
142
        "is_admin": True,
143
        "password1": "1402",
144
        "password2": "1402"
145
    })
146
    user_db = await UserService().sudo().create(user)
147
    admin_user = CurrentUser(**user_db.__dict__)
148

149
    yield admin_user
150
    await db_session.delete(user_db)
151
    await db_session.commit()
152

153

154
@pytest_asyncio.fixture
155
async def companies(db_session: AsyncSession, user_admin) -> Company:
156
    company1 = CompanyCreateScheme(title="Test company 1", currency='USD')
157
    company2 = CompanyCreateScheme(title="Test company 2", currency='RUB')
158
    company1_db = await CompanyService(user_admin).create(company1)
159
    company2_db = await CompanyService(user_admin).create(company2)
160
    yield [company1_db, company2_db]
161
    await db_session.delete(company1_db)
162
    await db_session.delete(company2_db)
163
    await db_session.commit()
164

165
@pytest_asyncio.fixture
166
async def stores(db_session: AsyncSession, user_admin, companies) -> Company:
167
    store1 = StoreCreateScheme(title="Store company 1", company_id=companies[0].id, address='addres 1')
168
    store2 = StoreCreateScheme(title="Store company 2",  company_id=companies[1].id, address='addres 1')
169
    store1_db = await StoreService(user_admin).create(store1)
170
    store2_db = await StoreService(user_admin).create(store2)
171
    yield [store1_db, store2_db]
172
    await db_session.delete(store1_db)
173
    await db_session.delete(store2_db)
174
    await db_session.commit()
175

176
@pytest_asyncio.fixture
177
async def product_categories(db_session: AsyncSession, user_admin, companies) -> Company:
178
    product_category1 = await ProductCategoryService(user_admin).create(ProductCategoryCreateScheme(**{
179
        'company_id': companies[0].id.__str__(),
180
        'title': 'Some Category 1'
181
    }))
182
    product_category2 = await ProductCategoryService(user_admin).create(ProductCategoryCreateScheme(**{
183
        'company_id': companies[1].id.__str__(),
184
        'title': 'Some Category 2'
185
    }))
186
    yield [product_category1, product_category2]
187
    await db_session.delete(product_category1)
188
    await db_session.delete(product_category2)
189
    await db_session.commit()
190

191
@pytest_asyncio.fixture
192
async def product_storage_types(db_session: AsyncSession, user_admin, companies) -> Company:
193
    product_storage_type1 = await ProductStorageTypeService(user_admin).create(ProductStorageTypeCreateScheme(**{
194
        'company_id': companies[0].id.__str__(),
195
        'title': 'Product Storage Type 1'
196
    }))
197
    product_storage_type2 = await ProductStorageTypeService(user_admin).create(ProductStorageTypeCreateScheme(**{
198
        'company_id': companies[1].id.__str__(),
199
        'title': 'Product Storage Type 2'
200
    }))
201
    yield [product_storage_type1, product_storage_type2]
202
    await db_session.delete(product_storage_type1)
203
    await db_session.delete(product_storage_type2)
204
    await db_session.commit()
205

206
@pytest_asyncio.fixture
207
async def uom_categories(db_session: AsyncSession, user_admin, companies) -> Company:
208
    uom_category1 = await UomCategoryService(user_admin).create(UomCategoryCreateScheme(**{
209
        'company_id': companies[0].id.__str__(),
210
        'title': 'Uom category 1'
211
    }))
212
    uom_category2 = await UomCategoryService(user_admin).create(UomCategoryCreateScheme(**{
213
        'company_id': companies[1].id.__str__(),
214
        'title': 'Uom category 2'
215
    }))
216
    yield [uom_category1, uom_category2]
217
    await db_session.delete(uom_category1)
218
    await db_session.delete(uom_category2)
219
    await db_session.commit()
220

221
@pytest_asyncio.fixture
222
async def uoms(db_session: AsyncSession, user_admin, companies, uom_categories) -> Company:
223
    uom1 = await UomService(user_admin).create(UomCreateScheme(**{
224
        'company_id': companies[0].id.__str__(),
225
        'title': 'ST',
226
        'category_id': uom_categories[0].id.__str__(),
227
        'type': UomType.STANDART,
228
        'ratio': 1,
229
        'precision': 1
230
    }))
231
    uom2 = await UomService(user_admin).create(UomCreateScheme(**{
232
        'company_id': companies[0].id.__str__(),
233
        'title': 'PAK10',
234
        'category_id': uom_categories[0].id.__str__(),
235
        'type': UomType.BIGGER,
236
        'ratio': 12,
237
        'precision': 0.5
238
    }))
239
    yield [uom1, uom2]
240
    await db_session.delete(uom1)
241
    await db_session.delete(uom2)
242
    await db_session.commit()
243

244
@pytest_asyncio.fixture
245
async def products(db_session: AsyncSession, user_admin, companies, uoms, product_storage_types, product_categories) -> Company:
246
    product1 = await ProductService(user_admin).create(ProductCreateScheme(**{
247
        'company_id': companies[0].id.__str__(),
248
        'title': 'Product 1',
249
        'description': 'Product Desc 1',
250
        'external_number': 'Product 1',
251
        'product_type': ProductType.STORABLE,
252
        'uom_id': uoms[0].id.__str__(),
253
        'product_category_id': product_categories[0].id.__str__(),
254
        'product_storage_type_id': product_storage_types[0].id.__str__(),
255
        'barcode_list': ['Product1 Barcode1', 'Product1 Barcode2'],
256
    }))
257
    product2 = await ProductService(user_admin).create(ProductCreateScheme(**{
258
        'company_id': companies[1].id.__str__(),
259
        'title': 'Product 1',
260
        'description': 'Product 2',
261
        'external_number': 'Product 2',
262
        'product_type': ProductType.STORABLE,
263
        'uom_id': uoms[1].id.__str__(),
264
        'product_category_id': product_categories[1].id.__str__(),
265
        'product_storage_type_id': product_storage_types[1].id.__str__(),
266
        'barcode_list': ['Product2 Barcode1', 'Product2 Barcode2'],
267
    }))
268
    yield [product1, product2]
269
    await db_session.delete(product1)
270
    await db_session.delete(product2)
271
    await db_session.commit()
272

273

274
@pytest_asyncio.fixture
275
async def location_types(db_session: AsyncSession, user_admin, companies) -> Company:
276
    location_type_partner = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
277
        'company_id': companies[0].id.__str__(),
278
        'title': 'PARTNER',
279
        'location_class': LocationClass.PARTNER
280
    }))
281
    location_type_place = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
282
        'company_id': companies[0].id.__str__(),
283
        'title': 'PLACE',
284
        'location_class': LocationClass.PLACE
285
    }))
286
    location_type_resource = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
287
        'company_id': companies[0].id.__str__(),
288
        'title': 'RESOURCE',
289
        'location_class': LocationClass.RESOURCE
290
    }))
291
    location_type_package = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
292
        'company_id': companies[0].id.__str__(),
293
        'title': 'PACKAGE',
294
        'location_class': LocationClass.PACKAGE
295
    }))
296
    location_type_zone = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
297
        'company_id': companies[0].id.__str__(),
298
        'title': 'ZONE',
299
        'location_class': LocationClass.ZONE
300
    }))
301
    location_type_lost = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
302
        'company_id': companies[0].id.__str__(),
303
        'title': 'LOST',
304
        'location_class': LocationClass.LOST
305
    }))
306
    location_type_inventory = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
307
        'company_id': companies[0].id.__str__(),
308
        'title': 'INVENTORY',
309
        'location_class': LocationClass.INVENTORY
310
    }))
311
    location_type_scrap = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
312
        'company_id': companies[0].id.__str__(),
313
        'title': 'SCRAP',
314
        'location_class': LocationClass.SCRAP
315
    }))
316
    location_type_scrapped = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
317
        'company_id': companies[0].id.__str__(),
318
        'title': 'SCRAPPED',
319
        'location_class': LocationClass.SCRAPPED
320
    }))
321
    location_type_buffer = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
322
        'company_id': companies[0].id.__str__(),
323
        'title': 'BUFFER',
324
        'location_class': LocationClass.BUFFER
325
    }))
326
    yield {
327
        'partner': location_type_partner,
328
        'place': location_type_place,
329
        'resource': location_type_resource,
330
        'package': location_type_package,
331
        'zone': location_type_zone,
332
        'lost': location_type_lost,
333
        'inventory': location_type_inventory,
334
        'scrap': location_type_scrap,
335
        'scrapped': location_type_scrapped,
336
        'buffer': location_type_buffer
337
    }
338
    await db_session.delete(location_type_partner)
339
    await db_session.delete(location_type_place)
340
    await db_session.delete(location_type_resource)
341
    await db_session.delete(location_type_package)
342
    await db_session.delete(location_type_zone)
343
    await db_session.delete(location_type_lost)
344
    await db_session.delete(location_type_inventory)
345
    await db_session.delete(location_type_scrap)
346
    await db_session.delete(location_type_scrapped)
347
    await db_session.delete(location_type_buffer)
348
    await db_session.commit()
349

350

351
@pytest_asyncio.fixture
352
async def locations(db_session: AsyncSession, user_admin, companies, stores, location_types, product_storage_types) -> dict:
353
    location_partner = await LocationService(user_admin).create(LocationCreateScheme(**{
354
        'company_id': companies[0].id.__str__(),
355
        'title': 'PARTNER',
356
        'store_id': stores[0].id.__str__(),
357
        #'parent_id': 'PARTNER',
358
        'is_active': True,
359
        'location_type_id': location_types['partner'].id.__str__(),
360
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
361
        #'partner_id': 'PARTNER',
362
        'is_homogeneity': False,
363
        'is_allow_create_package': True,
364
        'allowed_package_ids': [],
365
        'exclusive_package_ids': [],
366
        'allowed_order_type_ids': [],
367
        'exclusive_order_type_ids': [],
368
        'strategy': PutawayStrategy.FEFO
369

370
    }))
371
    location_place = await LocationService(user_admin).create(LocationCreateScheme(**{
372
        'company_id': companies[0].id.__str__(),
373
        'title': 'PLACE',
374
        'store_id': stores[0].id.__str__(),
375
        # 'parent_id': 'PARTNER',
376
        'is_active': True,
377
        'location_type_id': location_types['place'].id.__str__(),
378
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
379
        # 'partner_id': 'PARTNER',
380
        'is_homogeneity': False,
381
        'is_allow_create_package': True,
382
        'allowed_package_ids': [],
383
        'exclusive_package_ids': [],
384
        'allowed_order_type_ids': [],
385
        'exclusive_order_types_ids': [],
386
        'strategy': PutawayStrategy.FEFO
387
    }))
388
    location_resource = await LocationService(user_admin).create(LocationCreateScheme(**{
389
        'company_id': companies[0].id.__str__(),
390
        'title': 'RESOURCE',
391
        'store_id': stores[0].id.__str__(),
392
        # 'parent_id': 'PARTNER',
393
        'is_active': True,
394
        'location_type_id': location_types['resource'].id.__str__(),
395
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
396
        # 'partner_id': 'PARTNER',
397
        'is_homogeneity': False,
398
        'is_allow_create_package': True,
399
        'allowed_package_ids': [],
400
        'exclusive_package_ids': [],
401
        'allowed_order_type_ids': [],
402
        'exclusive_order_types_ids': [],
403
        'strategy': PutawayStrategy.FEFO
404
    }))
405
    location_package = await LocationService(user_admin).create(LocationCreateScheme(**{
406
        'company_id': companies[0].id.__str__(),
407
        'title': 'PACKAGE',
408
        'store_id': stores[0].id.__str__(),
409
        # 'parent_id': 'PARTNER',
410
        'is_active': True,
411
        'location_type_id': location_types['package'].id.__str__(),
412
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
413
        # 'partner_id': 'PARTNER',
414
        'is_homogeneity': False,
415
        'is_allow_create_package': True,
416
        'allowed_package_ids': [],
417
        'exclusive_package_ids': [],
418
        'allowed_order_type_ids': [],
419
        'exclusive_order_types_ids': [],
420
        'strategy': PutawayStrategy.FEFO
421
    }))
422
    location_zone = await LocationService(user_admin).create(LocationCreateScheme(**{
423
        'company_id': companies[0].id.__str__(),
424
        'title': 'ZONE',
425
        'store_id': stores[0].id.__str__(),
426
        # 'parent_id': 'PARTNER',
427
        'is_active': True,
428
        'location_type_id': location_types['zone'].id.__str__(),
429
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
430
        # 'partner_id': 'PARTNER',
431
        'is_homogeneity': False,
432
        'is_allow_create_package': True,
433
        'allowed_package_ids': [],
434
        'exclusive_package_ids': [],
435
        'allowed_order_type_ids': [],
436
        'exclusive_order_types_ids': [],
437
        'strategy': PutawayStrategy.FEFO
438
    }))
439
    location_lost = await LocationService(user_admin).create(LocationCreateScheme(**{
440
        'company_id': companies[0].id.__str__(),
441
        'title': 'LOST',
442
        'store_id': stores[0].id.__str__(),
443
        # 'parent_id': 'PARTNER',
444
        'is_active': True,
445
        'location_type_id': location_types['lost'].id.__str__(),
446
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
447
        # 'partner_id': 'PARTNER',
448
        'is_homogeneity': False,
449
        'is_allow_create_package': True,
450
        'allowed_package_ids': [],
451
        'exclusive_package_ids': [],
452
        'allowed_order_type_ids': [],
453
        'exclusive_order_types_ids': [],
454
        'strategy': PutawayStrategy.FEFO
455
    }))
456
    location_inventory = await LocationService(user_admin).create(LocationCreateScheme(**{
457
        'company_id': companies[0].id.__str__(),
458
        'title': 'LOST',
459
        'store_id': stores[0].id.__str__(),
460
        # 'parent_id': 'PARTNER',
461
        'is_active': True,
462
        'location_type_id': location_types['inventory'].id.__str__(),
463
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
464
        # 'partner_id': 'PARTNER',
465
        'is_homogeneity': False,
466
        'is_allow_create_package': True,
467
        'allowed_package_ids': [],
468
        'exclusive_package_ids': [],
469
        'allowed_order_type_ids': [],
470
        'exclusive_order_types_ids': [],
471
        'strategy': PutawayStrategy.FEFO
472
    }))
473
    location_scrap = await LocationService(user_admin).create(LocationCreateScheme(**{
474
        'company_id': companies[0].id.__str__(),
475
        'title': 'SCRAP',
476
        'store_id': stores[0].id.__str__(),
477
        # 'parent_id': 'PARTNER',
478
        'is_active': True,
479
        'location_type_id': location_types['scrap'].id.__str__(),
480
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
481
        # 'partner_id': 'PARTNER',
482
        'is_homogeneity': False,
483
        'is_allow_create_package': True,
484
        'allowed_package_ids': [],
485
        'exclusive_package_ids': [],
486
        'allowed_order_type_ids': [],
487
        'exclusive_order_types_ids': [],
488
        'strategy': PutawayStrategy.FEFO
489
    }))
490
    location_scrapped = await LocationService(user_admin).create(LocationCreateScheme(**{
491
        'company_id': companies[0].id.__str__(),
492
        'title': 'SCRAPPED',
493
        'store_id': stores[0].id.__str__(),
494
        # 'parent_id': 'PARTNER',
495
        'is_active': True,
496
        'location_type_id': location_types['scrapped'].id.__str__(),
497
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
498
        # 'partner_id': 'PARTNER',
499
        'is_homogeneity': False,
500
        'is_allow_create_package': True,
501
        'allowed_package_ids': [],
502
        'exclusive_package_ids': [],
503
        'allowed_order_type_ids': [],
504
        'exclusive_order_types_ids': [],
505
        'strategy': PutawayStrategy.FEFO
506
    }))
507
    location_buffer = await LocationService(user_admin).create(LocationCreateScheme(**{
508
        'company_id': companies[0].id.__str__(),
509
        'title': 'SCRAPPED',
510
        'store_id': stores[0].id.__str__(),
511
        # 'parent_id': 'PARTNER',
512
        'is_active': True,
513
        'location_type_id': location_types['buffer'].id.__str__(),
514
        'product_storage_type_ids': [i.id.__str__() for i in product_storage_types],
515
        # 'partner_id': 'PARTNER',
516
        'is_homogeneity': False,
517
        'is_allow_create_package': True,
518
        'allowed_package_ids': [],
519
        'exclusive_package_ids': [],
520
        'allowed_order_type_ids': [],
521
        'exclusive_order_types_ids': [],
522
        'strategy': PutawayStrategy.FEFO
523
    }))
524
    yield {
525
        'partner': location_partner,
526
        'place': location_place,
527
        'resource': location_resource,
528
        'package': location_package,
529
        'zone': location_zone,
530
        'lost': location_lost,
531
        'inventory': location_inventory,
532
        'scrap': location_scrap,
533
        'scrapped': location_scrapped,
534
        'buffer': location_buffer
535
    }
536
    await db_session.delete(location_partner)
537
    await db_session.delete(location_place)
538
    await db_session.delete(location_resource)
539
    await db_session.delete(location_package)
540
    await db_session.delete(location_zone)
541
    await db_session.delete(location_lost)
542
    await db_session.delete(location_inventory)
543
    await db_session.delete(location_scrap)
544
    await db_session.delete(location_scrapped)
545
    await db_session.delete(location_buffer)
546
    await db_session.commit()
547

548
@pytest_asyncio.fixture
549
async def lots(db_session: AsyncSession, user_admin, companies, products) -> Company:
550
    lot1 = await LotService(user_admin).create(LotCreateScheme(**{
551
        'company_id': companies[0].id.__str__(),
552
        'expiration_datetime': datetime.now().isoformat(),
553
        'product_id': products[0].id.__str__(),
554
        'external_number': '1000001',
555
    }))
556
    lot2 = await LotService(user_admin).create(LotCreateScheme(**{
557
        'company_id': companies[0].id.__str__(),
558
        'expiration_datetime': datetime.now().isoformat(),
559
        'product_id': products[0].id.__str__(),
560
        'external_number': '1000002',
561
    }))
562
    yield [lot1, lot2]
563
    await db_session.delete(lot1)
564
    await db_session.delete(lot2)
565
    await db_session.commit()
566

567
@pytest_asyncio.fixture
568
async def quants(db_session: AsyncSession, user_admin, companies, lots, products, stores, locations, uoms) -> Company:
569
    quant1 = await QuantService(user_admin).create(QuantCreateScheme(**{
570
        'company_id': companies[0].id.__str__(),
571
        'product_id': products[0].id.__str__(),
572
        'store_id': stores[0].id.__str__(),
573
        'location_id': locations['place'].id.__str__(),
574
        'lot_id': lots[0].id.__str__(),
575
        'quantity': 10.0,
576
        'reserved_quantity': 5,
577
        'expiration_datetime': datetime.now().isoformat(),
578
        'uom_id': uoms[0].id.__str__()
579
    }))
580
    quant2 = await QuantService(user_admin).create(QuantCreateScheme(**{
581
        'company_id': companies[0].id.__str__(),
582
        'product_id': products[0].id.__str__(),
583
        'store_id': stores[0].id.__str__(),
584
        'location_id': locations['zone'].id.__str__(),
585
        'lot_id': lots[1].id.__str__(),
586
        'quantity': 5,
587
        'reserved_quantity': 0,
588
        'expiration_datetime': datetime.now().isoformat(),
589
        'uom_id': uoms[0].id.__str__()
590
    }))
591
    yield [quant1, quant2]
592
    await db_session.delete(quant1)
593
    await db_session.delete(quant2)
594
    await db_session.commit()
595

596
@pytest_asyncio.fixture
597
async def order_types(db_session: AsyncSession, user_admin, companies, lots, products, stores, locations,uoms, token) -> Company:
598
    inbound_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
599
        'company_id': companies[0].id.__str__(),
600
        'prefix': 'IN',
601
        'title': 'Inbound Type',
602
        'order_class': 'incoming',
603
        'allowed_location_src_ids': [locations['partner'].id.__str__(), ],
604
        'exclusive_location_src_ids': None,
605
        'allowed_location_dest_ids': [locations['buffer'].id.__str__(), ],
606
        'exclusive_location_dest_ids': None,
607
        'backorder_order_type_id': None,
608
        'backorder_action_type': 'ask',
609
        'store_id': None,
610
        'partner_id': None,
611
        'reservation_method': 'at_confirm',
612
        'reservation_time_before': 0,
613
        #'allowed_package_ids': [locations['package'].id.__str__(), ],
614
        'exclusive_package_ids': None,
615
        'is_homogeneity': False,
616
        'is_allow_create_package': True,
617
        'can_create_order_manualy': True,
618
        'overdelivery': False,
619
        'created_by': token['user_admin']['user_id'].__str__(),
620
        'edited_by': token['user_admin']['user_id'].__str__(),
621
        'barcode': '2132132131231',
622
        'strategy': 'fefo',
623
    }))
624
    lost_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
625
        'company_id': companies[0].id.__str__(),
626
        'prefix': 'LO',
627
        'title': 'Lost Type',
628
        'order_class': 'incoming',
629
        'allowed_location_src_ids': None,
630
        'exclusive_location_src_ids': None,
631
        'allowed_location_dest_ids': [locations['lost'].id.__str__(), ],
632
        'exclusive_location_dest_ids': None,
633
        'backorder_order_type_id': None,
634
        'backorder_action_type': 'never',
635
        'store_id': None,
636
        'partner_id': None,
637
        'reservation_method': 'at_confirm',
638
        'reservation_time_before': 0,
639
        # 'allowed_package_ids': [locations['package'].id.__str__(), ],
640
        'exclusive_package_ids': None,
641
        'is_homogeneity': False,
642
        'is_allow_create_package': False,
643
        'can_create_order_manualy': True,
644
        'overdelivery': False,
645
        'created_by': token['user_admin']['user_id'].__str__(),
646
        'edited_by': token['user_admin']['user_id'].__str__(),
647
        'barcode': '2132132131231',
648
        'strategy': 'fefo',
649
    }))
650
    placement_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
651
        'company_id': companies[0].id.__str__(),
652
        'prefix': 'PL',
653
        'title': 'Placement Type',
654
        'order_class': 'internal',
655
        'allowed_location_src_ids': [locations['buffer'].id.__str__(), ],
656
        'exclusive_location_src_ids': None,
657
        'allowed_location_dest_ids': [locations['place'].id.__str__(), ],
658
        'exclusive_location_dest_ids': None,
659
        'backorder_order_type_id': lost_order_type.id.__str__(),
660
        'backorder_action_type': 'always',
661
        'store_id': None,
662
        'partner_id': None,
663
        'reservation_method': 'at_confirm',
664
        'reservation_time_before': 0,
665
        # 'allowed_package_ids': [locations['package'].id.__str__(), ],
666
        'exclusive_package_ids': None,
667
        'is_homogeneity': False,
668
        'is_allow_create_package': True,
669
        'can_create_order_manualy': True,
670
        'overdelivery': False,
671
        'created_by': token['user_admin']['user_id'].__str__(),
672
        'edited_by': token['user_admin']['user_id'].__str__(),
673
        'barcode': '2132132131231',
674
        'strategy': 'fefo',
675
    }))
676
    shipment_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
677
        'company_id': companies[0].id.__str__(),
678
        'prefix': 'SH',
679
        'title': 'Shipment Type',
680
        'order_class': 'outgoing',
681
        'allowed_location_src_ids': [locations['place'].id.__str__(), ] ,
682
        'exclusive_location_src_ids': None,
683
        'allowed_location_dest_ids': [locations['buffer'].id.__str__(), ],
684
        'exclusive_location_dest_ids': None,
685
        'backorder_order_type_id': None,
686
        'backorder_action_type': 'ask',
687
        'store_id': None,
688
        'partner_id': None,
689
        'reservation_method': 'at_confirm',
690
        'reservation_time_before': 0,
691
        # 'allowed_package_ids': [locations['package'].id.__str__(), ],
692
        'exclusive_package_ids': None,
693
        'is_homogeneity': False,
694
        'is_allow_create_package': True,
695
        'can_create_order_manualy': True,
696
        'overdelivery': False,
697
        'created_by': token['user_admin']['user_id'].__str__(),
698
        'edited_by': token['user_admin']['user_id'].__str__(),
699
        'barcode': '2132132131231',
700
        'strategy': 'fefo',
701
    }))
702

703
    yield {
704
        'inbound': inbound_order_type,
705
        'placement': placement_order_type,
706
        'lost': lost_order_type,
707
        'shipment': shipment_order_type
708
    }
709
    await db_session.delete(inbound_order_type)
710
    await db_session.delete(placement_order_type)
711
    await db_session.delete(lost_order_type)
712
    await db_session.delete(shipment_order_type)
713
    await db_session.commit()
714

715

716

717
@pytest_asyncio.fixture
718
async def roles(db_session: AsyncSession, companies, user_admin) -> User:
719
    permission_allow_list = [
720
        "user_create", 'user_edit', 'user_list', 'user_get', 'partner_create',
721
        'partner_edit', 'partner_list', 'partner_delete', 'partner_get',
722
        'company_create', 'company_edit', 'company_list', 'company_get',
723
        'uom_create', 'uom_edit', 'uom_list', 'uom_delete', 'uom_get']
724
    role_admin = RoleCreateScheme(title="admin", permission_allow_list=list(permits.keys()), company_id=companies[0].id)
725
    role_admin_db = await RoleService(user_admin).create(role_admin)
726
    role_support = RoleCreateScheme(title="support", permission_allow_list=permission_allow_list, company_id=companies[0].id,
727
                                    parent_ids=[role_admin_db.id])
728
    role_support_db = await RoleService(user_admin).create(role_support)
729
    yield {'admin': role_admin_db, 'support': role_support_db}
730
    await db_session.delete(role_admin_db)
731
    await db_session.delete(role_support_db)
732
    await db_session.commit()
733

734

735
@pytest_asyncio.fixture
736
async def users(db_session: AsyncSession, companies, roles, user_admin) -> User:
737
    company_admin = UserCreateScheme(**{
738
        "email": "company_admin@gmail.com",
739
        "nickname": "Admin vasya",
740
        "locale": "EN",
741
        "is_admin": False,
742
        "company_ids": [
743
            companies[0].id
744
        ],
745
        "role_ids": [
746
            roles.get('admin').id
747
        ],
748
        "password1": "1402",
749
        "password2": "1402"
750
    })
751
    company_support = UserCreateScheme(**{
752
        "email": "company_support@gmail.com",
753
        "nickname": "Support vasya",
754
        "is_admin": False,
755
        "locale": "RU",
756
        "company_ids": [
757
            companies[0].id
758
        ],
759
        "role_ids": [
760
            roles.get('support').id
761
        ],
762
        "password1": "1402",
763
        "password2": "1402"
764
    })
765
    company_admin_db = await UserService(user_admin).create(company_admin)
766
    company_support_db = await UserService(user_admin).create(company_support)
767
    yield {'company_admin': company_admin_db, 'company_support': company_support_db}
768
    await db_session.delete(company_admin_db)
769
    await db_session.delete(company_support_db)
770
    await db_session.commit()
771

772

773
@pytest_asyncio.fixture
774
async def token(db_session: AsyncSession, users: User, user_admin) -> dict[str, Any]:
775
    user_admin = await UserService(None).login('admin@admin.com', '1402')
776
    company_admin = await UserService(None).login(users.get('company_admin').email, users.get('company_admin').password)
777
    company_support = await UserService(None).login(users.get('company_support').email,
778
                                                    users.get('company_support').password)
779
    return {
780
        'user_admin': user_admin,
781
        'company_admin': company_admin,
782
        'company_support': company_support
783
    }
784

785

786

787
@pytest_asyncio.fixture
788
async def headers(token) -> dict:
789
    return {
790
        'superadmin': {'Authorization': token['user_admin']['token']},
791
        'company_admin': {'Authorization': token['company_admin']['token']},
792
        'company_support': {'Authorization': token['company_support']['token']}
793
    }
794

795

796
# @pytest.mark.asyncio
797
# async def test_health(async_client, headers, stores, product_categories, product_storage_types, uom_categories, uoms, products, locations, quants):
798
#     response = await async_client.get("/api/basic/health", headers=headers['superadmin'])
799
#     assert response.status_code == 200
800