2
from datetime import datetime
3
from typing import Type, Dict, Any, List
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
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
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()
53
@pytest.fixture(scope="session")
55
engine = create_async_engine(config.TEST_WRITER_DB_URL)
57
engine.sync_engine.dispose()
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",
66
db_name_base = config.DB_NAME_TEST
68
db_name = f'{db_name_base}_{i}'
70
async with create_db_engine.begin() as connection:
71
await connection.execute(
73
"drop database if exists {name};".format(
78
await connection.execute(
79
text("create database {name};".format(name=db_name)),
81
except Exception as ex:
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,
96
async with TestingSessionLocal(bind=connection) as session:
99
await session.rollback()
102
@pytest.fixture(scope="session")
103
def override_get_db(prepare_db, db_session: AsyncSession):
104
async def _override_get_db():
107
return _override_get_db
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):
115
elif isinstance(request, Request):
116
self.user = request.user
118
self.session = db_session
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:
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:
136
@pytest_asyncio.fixture
137
async def user_admin(db_session: AsyncSession) -> User:
138
user = UserCreateScheme(**{
139
"email": "admin@admin.com",
146
user_db = await UserService().sudo().create(user)
147
admin_user = CurrentUser(**user_db.__dict__)
150
await db_session.delete(user_db)
151
await db_session.commit()
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()
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()
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'
182
product_category2 = await ProductCategoryService(user_admin).create(ProductCategoryCreateScheme(**{
183
'company_id': companies[1].id.__str__(),
184
'title': 'Some Category 2'
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()
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'
197
product_storage_type2 = await ProductStorageTypeService(user_admin).create(ProductStorageTypeCreateScheme(**{
198
'company_id': companies[1].id.__str__(),
199
'title': 'Product Storage Type 2'
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()
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'
212
uom_category2 = await UomCategoryService(user_admin).create(UomCategoryCreateScheme(**{
213
'company_id': companies[1].id.__str__(),
214
'title': 'Uom category 2'
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()
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__(),
226
'category_id': uom_categories[0].id.__str__(),
227
'type': UomType.STANDART,
231
uom2 = await UomService(user_admin).create(UomCreateScheme(**{
232
'company_id': companies[0].id.__str__(),
234
'category_id': uom_categories[0].id.__str__(),
235
'type': UomType.BIGGER,
240
await db_session.delete(uom1)
241
await db_session.delete(uom2)
242
await db_session.commit()
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'],
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'],
268
yield [product1, product2]
269
await db_session.delete(product1)
270
await db_session.delete(product2)
271
await db_session.commit()
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__(),
279
'location_class': LocationClass.PARTNER
281
location_type_place = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
282
'company_id': companies[0].id.__str__(),
284
'location_class': LocationClass.PLACE
286
location_type_resource = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
287
'company_id': companies[0].id.__str__(),
289
'location_class': LocationClass.RESOURCE
291
location_type_package = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
292
'company_id': companies[0].id.__str__(),
294
'location_class': LocationClass.PACKAGE
296
location_type_zone = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
297
'company_id': companies[0].id.__str__(),
299
'location_class': LocationClass.ZONE
301
location_type_lost = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
302
'company_id': companies[0].id.__str__(),
304
'location_class': LocationClass.LOST
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
311
location_type_scrap = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
312
'company_id': companies[0].id.__str__(),
314
'location_class': LocationClass.SCRAP
316
location_type_scrapped = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
317
'company_id': companies[0].id.__str__(),
319
'location_class': LocationClass.SCRAPPED
321
location_type_buffer = await LocationTypeService(user_admin).create(LocationTypeCreateScheme(**{
322
'company_id': companies[0].id.__str__(),
324
'location_class': LocationClass.BUFFER
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
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()
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__(),
356
'store_id': stores[0].id.__str__(),
357
#'parent_id': 'PARTNER',
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
371
location_place = await LocationService(user_admin).create(LocationCreateScheme(**{
372
'company_id': companies[0].id.__str__(),
374
'store_id': stores[0].id.__str__(),
375
# 'parent_id': 'PARTNER',
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
388
location_resource = await LocationService(user_admin).create(LocationCreateScheme(**{
389
'company_id': companies[0].id.__str__(),
391
'store_id': stores[0].id.__str__(),
392
# 'parent_id': 'PARTNER',
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
405
location_package = await LocationService(user_admin).create(LocationCreateScheme(**{
406
'company_id': companies[0].id.__str__(),
408
'store_id': stores[0].id.__str__(),
409
# 'parent_id': 'PARTNER',
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
422
location_zone = await LocationService(user_admin).create(LocationCreateScheme(**{
423
'company_id': companies[0].id.__str__(),
425
'store_id': stores[0].id.__str__(),
426
# 'parent_id': 'PARTNER',
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
439
location_lost = await LocationService(user_admin).create(LocationCreateScheme(**{
440
'company_id': companies[0].id.__str__(),
442
'store_id': stores[0].id.__str__(),
443
# 'parent_id': 'PARTNER',
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
456
location_inventory = await LocationService(user_admin).create(LocationCreateScheme(**{
457
'company_id': companies[0].id.__str__(),
459
'store_id': stores[0].id.__str__(),
460
# 'parent_id': 'PARTNER',
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
473
location_scrap = await LocationService(user_admin).create(LocationCreateScheme(**{
474
'company_id': companies[0].id.__str__(),
476
'store_id': stores[0].id.__str__(),
477
# 'parent_id': 'PARTNER',
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
490
location_scrapped = await LocationService(user_admin).create(LocationCreateScheme(**{
491
'company_id': companies[0].id.__str__(),
493
'store_id': stores[0].id.__str__(),
494
# 'parent_id': 'PARTNER',
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
507
location_buffer = await LocationService(user_admin).create(LocationCreateScheme(**{
508
'company_id': companies[0].id.__str__(),
510
'store_id': stores[0].id.__str__(),
511
# 'parent_id': 'PARTNER',
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
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
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()
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',
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',
563
await db_session.delete(lot1)
564
await db_session.delete(lot2)
565
await db_session.commit()
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__(),
576
'reserved_quantity': 5,
577
'expiration_datetime': datetime.now().isoformat(),
578
'uom_id': uoms[0].id.__str__()
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__(),
587
'reserved_quantity': 0,
588
'expiration_datetime': datetime.now().isoformat(),
589
'uom_id': uoms[0].id.__str__()
591
yield [quant1, quant2]
592
await db_session.delete(quant1)
593
await db_session.delete(quant2)
594
await db_session.commit()
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__(),
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',
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',
624
lost_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
625
'company_id': companies[0].id.__str__(),
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',
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',
650
placement_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
651
'company_id': companies[0].id.__str__(),
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',
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',
676
shipment_order_type = await OrderTypeService(user_admin).create(OrderTypeCreateScheme(**{
677
'company_id': companies[0].id.__str__(),
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',
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',
704
'inbound': inbound_order_type,
705
'placement': placement_order_type,
706
'lost': lost_order_type,
707
'shipment': shipment_order_type
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()
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()
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",
746
roles.get('admin').id
751
company_support = UserCreateScheme(**{
752
"email": "company_support@gmail.com",
753
"nickname": "Support vasya",
760
roles.get('support').id
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()
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)
780
'user_admin': user_admin,
781
'company_admin': company_admin,
782
'company_support': company_support
787
@pytest_asyncio.fixture
788
async def headers(token) -> dict:
790
'superadmin': {'Authorization': token['user_admin']['token']},
791
'company_admin': {'Authorization': token['company_admin']['token']},
792
'company_support': {'Authorization': token['company_support']['token']}
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