Coverage for tests/tests_api/fixtures.py: 44%

1import datetime 

2 

3import pytest_asyncio 

4from mongodb import ( 

5 buf_col, 

6 clients_col, 

7 orders_col, 

8 orgs_col, 

9 subsidiaries_col, 

10 trash_orders_col, 

11) 

12from sotrans_models.models.misc.client import ClientCreateModel, ClientDBModel 

13from sotrans_models.models.orders.order import OrderDBModel, OrderStatus 

14from utils.dt_utils import get_current_datetime 

15 

16from ..conftest import tmp_org, tmp_subsidiary, tmp_user 

17from ..random_model import generate_models_json 

18 

19 

20@pytest_asyncio.fixture 

21async def tmp_client(): 

22 model_dump = generate_models_json(ClientCreateModel) 

23 created = await clients_col.create(model_dump) 

24 client = ClientDBModel(**created) 

25 yield client 

26 await clients_col.delete_by_id(client.id) 

27 

28 

29@pytest_asyncio.fixture 

30async def tmp_buffer_order(tmp_client): 

31 model_dump = generate_models_json( 

32 OrderDBModel, 

33 { 

34 "status": OrderStatus.buffer.value, 

35 "client": tmp_client.model_dump(), 

36 }, 

37 ) 

38 created = await buf_col.create(model_dump) 

39 yield OrderDBModel(**created) 

40 

41 

42@pytest_asyncio.fixture 

43async def tmp_trash_order(tmp_client): 

44 model_dump = generate_models_json( 

45 OrderDBModel, 

46 { 

47 "status": OrderStatus.archived.value, 

48 "client": tmp_client.model_dump(), 

49 }, 

50 ) 

51 created = await trash_orders_col.create(model_dump) 

52 yield OrderDBModel(**created) 

53 

54 

55@pytest_asyncio.fixture 

56async def tmp_confirmed_order(tmp_subsidiary, tmp_client): 

57 model_dump = generate_models_json( 

58 OrderDBModel, 

59 { 

60 "status": OrderStatus.confirmed.value, 

61 "assigned": { 

62 "company": {"subsidiary": tmp_subsidiary.model_dump()} 

63 }, 

64 "client": tmp_client.model_dump(), 

65 }, 

66 ) 

67 created = await orders_col.create(model_dump) 

68 yield OrderDBModel(**created) 

69 

70 

71@pytest_asyncio.fixture 

72async def tmp_exchange_order(tmp_subsidiary, tmp_client): 

73 model_dump = generate_models_json( 

74 OrderDBModel, 

75 { 

76 "status": OrderStatus.exchange.value, 

77 "start_price": 20_000, 

78 "rate_step": 1_000, 

79 "client": tmp_client.model_dump(), 

80 "auction_end_time": get_current_datetime() 

81 + datetime.timedelta(days=10), 

82 "on_exchange_from": None, 

83 "assigned": { 

84 "company": { 

85 "subsidiary": tmp_subsidiary.model_dump(format_ids=False) 

86 } 

87 }, 

88 }, 

89 ) 

90 created = await orders_col.create(model_dump) 

91 yield OrderDBModel(**created) 

92 

93 

94@pytest_asyncio.fixture 

95async def tmp_exchange_duple(tmp_client): 

96 model_dump = generate_models_json( 

97 OrderDBModel, 

98 { 

99 "status": OrderStatus.exchange.value, 

100 "start_price": 20_000, 

101 "rate_step": 1_000, 

102 "auction_end_time": get_current_datetime() 

103 + datetime.timedelta(days=10), 

104 "client": tmp_client.model_dump(), 

105 }, 

106 ) 

107 created = await orders_col.create(model_dump) 

108 yield OrderDBModel(**created) 

109 

110 

111@pytest_asyncio.fixture 

112async def update_org_with_employee(tmp_org, tmp_user): 

113 await orgs_col.collection.update_one( 

114 {"id": tmp_org.id}, {"$set": {"employees": [tmp_user.model_dump()]}} 

115 ) 

116 

117 

118@pytest_asyncio.fixture 

119async def update_subsidiary_with_employee(tmp_subsidiary, tmp_user): 

120 await subsidiaries_col.update_by_id( 

121 tmp_subsidiary.id, {"employees": [tmp_user.model_dump()]} 

122 ) 

123 yield tmp_subsidiary 

124 

125 

126@pytest_asyncio.fixture 

127async def tmp_appointment_order(tmp_subsidiary, tmp_client): 

128 tmp_order = generate_models_json( 

129 OrderDBModel, 

130 { 

131 "status": OrderStatus.appointment.value, 

132 "assigned": { 

133 "company": {"subsidiary": tmp_subsidiary.model_dump()} 

134 }, 

135 "client": tmp_client.model_dump(), 

136 }, 

137 ) 

138 created = await orders_col.create(tmp_order) 

139 order = OrderDBModel(**created) 

140 yield order 

141 await orders_col.delete_by_id(order.id) 

142 

143 

144@pytest_asyncio.fixture 

145async def tmp_canceled_order(tmp_subsidiary, tmp_client): 

146 tmp_order = generate_models_json( 

147 OrderDBModel, 

148 { 

149 "status": OrderStatus.canceled.value, 

150 "assigned": { 

151 "company": {"subsidiary": tmp_subsidiary.model_dump()} 

152 }, 

153 "client": tmp_client.model_dump(), 

154 }, 

155 ) 

156 created = await orders_col.create(tmp_order) 

157 order = OrderDBModel(**created) 

158 yield order 

159 await orders_col.delete_by_id(order.id) 

160 

161 

162@pytest_asyncio.fixture 

163async def tmp_unverified_order(tmp_org, tmp_subsidiary, tmp_client): 

164 tmp_order = generate_models_json( 

165 OrderDBModel, 

166 { 

167 "status": OrderStatus.unverified.value, 

168 "carrier": tmp_org.model_dump(), 

169 "assigned": { 

170 "company": {"subsidiary": tmp_subsidiary.model_dump()} 

171 }, 

172 "client": tmp_client.model_dump(), 

173 }, 

174 ) 

175 created = await orders_col.create(tmp_order) 

176 order = OrderDBModel(**created) 

177 yield order 

178 await orders_col.delete_by_id(order.id)