Coverage for chatgpt_proxy / tests / test_api.py: 99%

1# MIT License 

2# 

3# Copyright (c) 2025 Tuomo Kriikkula 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy 

6# of this software and associated documentation files (the "Software"), to deal 

7# in the Software without restriction, including without limitation the rights 

8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

9# copies of the Software, and to permit persons to whom the Software is 

10# furnished to do so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in all 

13# copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 

21# SOFTWARE. 

22 

23import asyncio 

24import datetime 

25import hashlib 

26import ipaddress 

27import logging 

28import os 

29from typing import AsyncGenerator 

30 

31import asyncpg 

32import httpx 

33import jwt 

34import nest_asyncio 

35import openai.types.responses as openai_responses 

36import pytest 

37import pytest_asyncio 

38import respx 

39from pytest_loguru.plugin import caplog # noqa: F401 

40from sanic.log import access_logger as sanic_access_logger 

41from sanic.log import logger as sanic_logger 

42from sanic_testing.reusable import ReusableClient 

43 

44from chatgpt_proxy.tests import setup # noqa: E402 

45 

46setup.common_test_setup() 

47 

48# noinspection PyUnresolvedReferences 

49import chatgpt_proxy # noqa: E402 

50from chatgpt_proxy import auth # noqa: E402 

51from chatgpt_proxy.app import app # noqa: E402 

52from chatgpt_proxy.app import game_id_length # noqa: E402 

53from chatgpt_proxy.app import make_api_v1_app # noqa: E402 

54from chatgpt_proxy.app import max_ast_literal_eval_size # noqa: E402 

55from chatgpt_proxy.app import openai_model # noqa: E402 

56from chatgpt_proxy.cache import app_cache # noqa: E402 

57from chatgpt_proxy.db import models # noqa: E402 

58from chatgpt_proxy.db import pool_acquire # noqa: E402 

59from chatgpt_proxy.db import queries # noqa: E402 

60from chatgpt_proxy.db.models import SayType # noqa: E402 

61from chatgpt_proxy.db.models import Team # noqa: E402 

62from chatgpt_proxy.log import logger # noqa: E402 

63from chatgpt_proxy.tests.client import SpoofedSanicASGITestClient # noqa: E402 

64from chatgpt_proxy.tests.monkey_patch import monkey_patch_sanic_testing # noqa: E402 

65from chatgpt_proxy.tests.setup import retry_context # noqa: E402 

66from chatgpt_proxy.types import App # noqa: E402 

67from chatgpt_proxy.utils import utcnow # noqa: E402 

68 

69logger.level("DEBUG") 

70 

71_asgi_host = "127.0.0.1" 

72 

73monkey_patch_sanic_testing( 

74 asgi_host=_asgi_host, 

75) 

76 

77_db_timeout = setup.default_test_db_timeout 

78 

79_game_server_address = ipaddress.IPv4Address("127.0.0.1") 

80_game_server_port = 7777 

81_now = utcnow() 

82_iat = _now 

83_exp = _now + datetime.timedelta(hours=12) 

84_token = jwt.encode( 

85 key=setup.test_sanic_secret, 

86 algorithm="HS256", 

87 payload={ 

88 "iss": auth.jwt_issuer, 

89 "aud": auth.jwt_audience, 

90 "sub": f"{_game_server_address}:{_game_server_port}", 

91 "iat": int(_iat.timestamp()), 

92 "exp": int(_exp.timestamp()), 

93 }, 

94) 

95_token_sha256 = hashlib.sha256(_token.encode()).digest() 

96_headers: dict[str, str] = { 

97 "Authorization": f"Bearer {_token}", 

98} 

99 

100_token_bad_extra_metadata = jwt.encode( 

101 key=setup.test_sanic_secret, 

102 algorithm="HS256", 

103 payload={ 

104 "iss": auth.jwt_issuer, 

105 "aud": auth.jwt_audience, 

106 "sub": f"{_game_server_address}:{_game_server_port}", 

107 "iat": int(_iat.timestamp()), 

108 "exp": int(_exp.timestamp()), 

109 "whatthefuck": "hmm?", 

110 }, 

111) 

112 

113_forbidden_game_server_address = ipaddress.IPv4Address("88.99.12.1") 

114_forbidden_game_server_port = 6969 

115_token_for_forbidden_server = jwt.encode( 

116 key=setup.test_sanic_secret, 

117 algorithm="HS256", 

118 payload={ 

119 "iss": auth.jwt_issuer, 

120 "aud": auth.jwt_audience, 

121 "sub": f"{_forbidden_game_server_address}:{_forbidden_game_server_port}", 

122 "iat": int(_iat.timestamp()), 

123 "exp": int(_exp.timestamp()), 

124 }, 

125) 

126_token_for_forbidden_server_sha256 = hashlib.sha256( 

127 _token_for_forbidden_server.encode()).digest() 

128 

129_steam_web_api_get_server_list_dummy_filter = ( 

130 f"\\gamedir\\rs2\\gameaddr\\{_game_server_address}:{_game_server_port}") 

131 

132app.config.ACCESS_LOG = True 

133app.config.OAS = False 

134app.config.OAS_AUTODOC = False 

135sanic_logger.setLevel(logging.DEBUG) 

136sanic_access_logger.setLevel(logging.DEBUG) 

137 

138# TODO: maybe just use namedtuple? 

139ApiFixtureTuple = tuple[ 

140 App, 

141 ReusableClient, 

142 respx.MockRouter, 

143 respx.MockRouter, 

144 asyncpg.Connection, 

145] 

146 

147 

148def patch_openai_response_output_text( 

149 mock_router: respx.MockRouter, 

150 output_text: str, 

151 method: str, 

152 status_code: int = 200, 

153): 

154 response = openai_responses.Response( 

155 id="testing_0", 

156 model=openai_model, 

157 created_at=utcnow().timestamp(), 

158 object="response", 

159 error=None, 

160 instructions=None, 

161 parallel_tool_calls=False, 

162 tool_choice="auto", 

163 tools=[], 

164 output=[ 

165 openai_responses.ResponseOutputMessage( 

166 id="msg_0_testing_0", 

167 content=[ 

168 openai_responses.ResponseOutputText( 

169 annotations=[], 

170 text=output_text, 

171 type="output_text", 

172 ), 

173 ], 

174 role="assistant", 

175 status="completed", 

176 type="message", 

177 ), 

178 ], 

179 ) 

180 

181 meth = getattr(mock_router, method) 

182 meth("/v1/responses").mock( 

183 return_value=httpx.Response( 

184 status_code=status_code, 

185 json=response.model_dump(mode="json"), 

186 )) 

187 

188 

189@pytest_asyncio.fixture 

190async def api_fixture( 

191) -> AsyncGenerator[ApiFixtureTuple]: 

192 loop = asyncio.get_running_loop() 

193 nest_asyncio.apply(loop=loop) 

194 

195 db_fixture_pool = await asyncpg.create_pool( 

196 dsn=setup.db_base_url, 

197 min_size=1, 

198 max_size=1, 

199 timeout=_db_timeout, 

200 loop=loop, 

201 ) 

202 

203 async with pool_acquire(db_fixture_pool, timeout=_db_timeout) as conn: 

204 await setup.drop_test_db(conn, timeout=_db_timeout) 

205 await setup.create_test_db(conn, timeout=_db_timeout) 

206 

207 test_db_pool = await asyncpg.create_pool( 

208 dsn=setup.db_test_url, 

209 min_size=1, 

210 max_size=1, 

211 timeout=_db_timeout, 

212 loop=loop, 

213 ) 

214 

215 async with pool_acquire( 

216 test_db_pool, 

217 timeout=_db_timeout, 

218 ) as conn: 

219 async with conn.transaction(): 

220 await setup.initialize_test_db(conn, timeout=_db_timeout) 

221 await setup.seed_test_db(conn, timeout=_db_timeout) 

222 

223 await queries.insert_game_server_api_key( 

224 conn=conn, 

225 issued_at=_iat, 

226 expires_at=_exp, 

227 token_hash=_token_sha256, 

228 game_server_address=_game_server_address, 

229 game_server_port=_game_server_port, 

230 name="pytest API key", 

231 ) 

232 await queries.insert_game_server_api_key( 

233 conn=conn, 

234 issued_at=_iat, 

235 expires_at=_exp, 

236 token_hash=_token_for_forbidden_server_sha256, 

237 game_server_address=_forbidden_game_server_address, 

238 game_server_port=_forbidden_game_server_port, 

239 name="pytest API key (forbidden game server)", 

240 ) 

241 

242 app.asgi_client.headers = _headers 

243 app.asgi_client.loop = loop 

244 

245 with (respx.MockRouter( 

246 base_url="https://api.openai.com/", 

247 assert_all_called=False, 

248 ) as openai_mock_router, 

249 respx.MockRouter( 

250 base_url="https://api.steampowered.com", 

251 assert_all_called=False, 

252 ) as steam_web_api_mock_router 

253 ): 

254 openai_mock_router.route(host=_asgi_host).pass_through() 

255 patch_openai_response_output_text( 

256 mock_router=openai_mock_router, 

257 output_text="This is a mocked test message!", 

258 method="post", 

259 ) 

260 

261 steam_web_api_mock_router.route(host=_asgi_host).pass_through() 

262 steam_web_api_mock_router.get( 

263 "IGameServersService/GetServerList/v1/", 

264 params={ 

265 "key": setup.steam_web_api_key, 

266 "filter": _steam_web_api_get_server_list_dummy_filter, 

267 } 

268 ).mock( 

269 return_value=httpx.Response( 

270 status_code=200, 

271 json={ 

272 "response": { 

273 "servers": [ 

274 { 

275 "addr": "127.0.0.1:27015", 

276 "gameport": 7777, 

277 "steamid": "43215698745632158", 

278 "name": "Dummy Server for pytest", 

279 "appid": 418460, 

280 "gamedir": "RS2", 

281 "version": "1094", 

282 "product": "RS2", 

283 "region": 255, 

284 "players": 3, 

285 "max_players": 64, 

286 "bots": 0, 

287 "map": "VNSK-Riverbed", 

288 "secure": True, 

289 "dedicated": True, 

290 "os": "w", 

291 "gametype": "does_not_matter" 

292 }, 

293 ], 

294 }, 

295 }, 

296 )) 

297 

298 def retry_cb(exc: Exception, retry: int) -> None: 

299 print("#" * 500) # TODO: remove me! 

300 logger.info( 

301 "reusable_client: retry attempt {}: error: {}: {}", 

302 retry, type(exc).__name__, exc, 

303 ) 

304 

305 # NOTE: can't reuse the same app for ReusableClient! 

306 reusable_app = make_api_v1_app( 

307 "ChatGPTProxy-Reusable", 

308 ) 

309 

310 def client_builder() -> ReusableClient: 

311 return ReusableClient( 

312 reusable_app, 

313 host=_asgi_host, 

314 loop=loop, 

315 client_kwargs={ 

316 "headers": _headers, 

317 } 

318 ) 

319 

320 with retry_context( 

321 client_builder, 

322 retries=5, 

323 delay=datetime.timedelta(milliseconds=500), 

324 exc_types=(PermissionError,), 

325 retry_cb=retry_cb, 

326 ) as reusable_client: 

327 yield app, reusable_client, openai_mock_router, steam_web_api_mock_router, conn 

328 

329 async with pool_acquire(db_fixture_pool, timeout=_db_timeout) as conn: 

330 await setup.drop_test_db(conn, timeout=_db_timeout) 

331 

332 await db_fixture_pool.close() 

333 await test_db_pool.close() 

334 

335 

336@pytest.mark.asyncio 

337async def test_api_v1_post_game(api_fixture, caplog) -> None: 

338 caplog.set_level(logging.DEBUG) 

339 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

340 

341 data = "VNTE-TestSuite\nTest Suite\n7777" 

342 req, resp = reusable_client.post("/api/v1/game", data=data) 

343 assert resp.status == 201 

344 

345 game_id, greeting = resp.text.split("\n") 

346 assert len(game_id) == game_id_length * 2 # Num bytes as hex string. 

347 

348 req, resp = reusable_client.get(f"/api/v1/game/{game_id}") 

349 assert resp.status == 200 

350 game = resp.json 

351 assert game 

352 

353 # Empty data. 

354 data = "" 

355 req, resp = reusable_client.post("/api/v1/game", data=data) 

356 assert resp.status == 400 

357 

358 # Bad data. 

359 data = "dsflkjgjknoe8923u58u234r02opkwepkf\n\r" 

360 req, resp = reusable_client.post("/api/v1/game", data=data) 

361 assert resp.status == 400 

362 

363 

364@pytest.mark.asyncio 

365async def test_api_v1_put_game(api_fixture, caplog) -> None: 

366 caplog.set_level(logging.DEBUG) 

367 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

368 

369 # Valid request. 

370 world_time = 548.8584 

371 data = f"{world_time}" 

372 req, resp = reusable_client.put("/api/v1/game/first_game", data=data) 

373 assert resp.status == 204 

374 

375 # Bad data -> 400. 

376 bad_world_time = "this is not a float" 

377 data = f"{bad_world_time}" 

378 req, resp = reusable_client.put("/api/v1/game/first_game", data=data) 

379 assert resp.status == 400 

380 

381 # Non-existent game. 

382 data = "this doesn't matter in this case!" 

383 req, resp = reusable_client.put("/api/v1/game/asdasdasd1243", data=data) 

384 assert resp.status == 404 

385 

386 

387@pytest.mark.asyncio 

388async def test_api_v1_post_game_invalid_token(api_fixture, caplog) -> None: 

389 caplog.set_level(logging.DEBUG) 

390 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

391 

392 # No token. 

393 logger.info("testing no token") 

394 data = "VNTE-TestSuite\n7777" 

395 req, resp = reusable_client.post( 

396 "/api/v1/game", 

397 data=data, 

398 headers={ 

399 "Authorization": "", 

400 }, 

401 ) 

402 assert resp.status == 401 

403 

404 # Bad token (garbage data). 

405 logger.info("testing bad token") 

406 data = "VNTE-TestSuite\n7777" 

407 req, resp = reusable_client.post( 

408 "/api/v1/game", 

409 data=data, 

410 headers={ 

411 "Authorization": "Bearer aasddsadasasdadsadsdsaasdasdads", 

412 }, 

413 ) 

414 assert resp.status == 401 

415 

416 # Good token with unwanted extra claims added 

417 # -> token hash check mismatch with database-stored hash. 

418 logger.info("testing token with extra metadata") 

419 data = "VNTE-TestSuite\n7777" 

420 req, resp = reusable_client.post( 

421 "/api/v1/game", 

422 data=data, 

423 headers={ 

424 "Authorization": f"Bearer {_token_bad_extra_metadata}", 

425 }, 

426 ) 

427 assert resp.status == 401 

428 

429 logger.info("testing Steam not recognizing the dedicated server") 

430 with steam_mock_router: 

431 steam_mock_router.get( 

432 "IGameServersService/GetServerList/v1/", 

433 params={ 

434 "key": setup.steam_web_api_key, 

435 "filter": _steam_web_api_get_server_list_dummy_filter, 

436 } 

437 ).mock( 

438 return_value=httpx.Response( 

439 status_code=200, 

440 json={ 

441 "response": { 

442 "servers": [], 

443 }, 

444 }, 

445 )) 

446 data = "VNTE-WhatTheFuckBro\n7777" 

447 req_, resp_ = reusable_client.post("/api/v1/game", data=data) 

448 assert resp_.status == 401, resp_.body 

449 

450 logger.info("testing Steam API returning garbage") 

451 with steam_mock_router: 

452 steam_mock_router.get( 

453 "IGameServersService/GetServerList/v1/", 

454 params={ 

455 "key": setup.steam_web_api_key, 

456 "filter": _steam_web_api_get_server_list_dummy_filter, 

457 } 

458 ).mock( 

459 return_value=httpx.Response( 

460 status_code=200, 

461 json={ 

462 "blasdalsd": { 

463 "xorvors": [], 

464 }, 

465 }, 

466 )) 

467 data = "VNTE-WhatTheFuckBro\n7777" 

468 req, resp = reusable_client.post("/api/v1/game", data=data) 

469 assert resp.status == 401 

470 

471 logger.info("testing token meant for another IP address (but the token exists in the DB)") 

472 data = "VNTE-ThisDoesntMatterLol\n7777" 

473 req, resp = reusable_client.post( 

474 "/api/v1/game", 

475 data=data, 

476 headers={ 

477 "Authorization": f"Bearer {_token_for_forbidden_server}", 

478 }, 

479 ) 

480 assert resp.status == 401 

481 

482 logger.info("testing token meant for another IP address") 

483 spoofed_asgi_client = SpoofedSanicASGITestClient( 

484 app=api_app, 

485 client_ip="6.0.28.175", 

486 ) 

487 somebody_elses_token = jwt.encode( 

488 key=setup.test_sanic_secret, 

489 algorithm="HS256", 

490 payload={ 

491 "iss": auth.jwt_issuer, 

492 "aud": auth.jwt_audience, 

493 "sub": "6.0.28.175:55555", # Also, this does not exist in the DB. 

494 "iat": int(_iat.timestamp()), 

495 "exp": int(_exp.timestamp()), 

496 }, 

497 ) 

498 spoofed_asgi_client.headers = { 

499 "Authorization": f"Bearer {somebody_elses_token}", 

500 } 

501 data = "VNTE-TestSuite\n55555" 

502 # noinspection PyTypeChecker 

503 req, resp = await spoofed_asgi_client.post("/api/v1/game", data=data) 

504 assert resp.status == 401 

505 

506 

507@pytest.mark.asyncio 

508async def test_api_v1_post_game_chat_message(api_fixture, caplog) -> None: 

509 # TODO: maybe just parametrize this test. 

510 

511 caplog.set_level(logging.DEBUG) 

512 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

513 

514 path = "/api/v1/game/first_game/chat_message" 

515 data = models.GameChatMessage( 

516 id=0, 

517 sender_name="my name is dog69", 

518 sender_team=Team.North, 

519 channel=SayType.ALL, 

520 message="this is the actual message!", 

521 game_id="first_game", 

522 send_time=utcnow(), 

523 ).wire_format() 

524 req, resp = reusable_client.post(path, data=data) 

525 assert resp.status == 204 

526 

527 path_404 = "/api/v1/game/THIS_GAME_DOES_NOT_EXIST/chat_message" 

528 data = "my name is dog69\n0\n0\nthis is the actual message!" 

529 req, resp = reusable_client.post(path_404, data=data) 

530 assert resp.status == 404 

531 

532 path_forbidden = "/api/v1/game/game_from_forbidden_server/chat_message" 

533 data = "my name is dog69\n0\n0\nthis is the actual message!" 

534 req, resp = reusable_client.post(path_forbidden, data=data) 

535 assert resp.status == 401 

536 

537 path = "/api/v1/game/first_game/chat_message" 

538 invalid_data = "dsfsdsfdsffdsfsdsdf" 

539 req, resp = reusable_client.post(path, data=invalid_data) 

540 assert resp.status == 400 

541 

542 path = "/api/v1/game/first_game/chat_message" 

543 empty_data = "" 

544 req, resp = reusable_client.post(path, data=empty_data) 

545 assert resp.status == 400 

546 

547 path = "/api/v1/game/first_game/chat_message" 

548 req, resp = reusable_client.post(path) # No data. 

549 assert resp.status == 400 

550 

551 # Steam Web API key is not set -> the result should still be the same, 

552 # only with a warning logged, but don't bother asserting the log message. 

553 # Run this check for coverage. 

554 try: 

555 # Make sure there aren't any cached Steam Web API results. 

556 await app_cache.clear() 

557 del os.environ["STEAM_WEB_API_KEY"] 

558 chatgpt_proxy.auth.load_config() 

559 path = "/api/v1/game/first_game/chat_message" 

560 data = "my name is dog69\n0\n0\nthis is the actual message!" 

561 req, resp = reusable_client.post(path, data=data) 

562 assert resp.status == 204 

563 finally: 

564 os.environ["STEAM_WEB_API_KEY"] = setup.steam_web_api_key 

565 chatgpt_proxy.auth.load_config() 

566 

567 num_steam_web_api_queries = await queries.select_steam_web_api_queries( 

568 conn=db_conn, 

569 ) 

570 assert num_steam_web_api_queries > 0 

571 

572 

573# TODO: make a dedicated test suite for "end-to-end" tests with 

574# a 'real' production setup? 

575# @pytest.mark.asyncio 

576# async def test_database_maintenance(api_fixture, caplog) -> None: 

577# pass 

578 

579 

580@pytest.mark.asyncio 

581async def test_api_v1_put_delete_game_player(api_fixture, caplog) -> None: 

582 caplog.set_level(logging.DEBUG) 

583 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

584 

585 # PUT a new player -> should be 201 CREATED. 

586 new_player_id = 69696 

587 path = f"/api/v1/game/first_game/player/{new_player_id}" 

588 data = "Bob\n1\n-50" 

589 req, resp = reusable_client.put(path, data=data) 

590 assert resp.status == 201 

591 player = await queries.select_game_player( 

592 conn=db_conn, 

593 game_id="first_game", 

594 player_id=new_player_id, 

595 ) 

596 assert player == models.GamePlayer( 

597 game_id="first_game", 

598 id=new_player_id, 

599 name="Bob", 

600 team=Team.South, 

601 score=-50, 

602 ) 

603 

604 # PUT existing player -> should be 204 NO CONTENT. 

605 data = models.GamePlayer( 

606 game_id="first_game", 

607 id=new_player_id, 

608 name="Bob", 

609 team=Team.South, 

610 score=6969, 

611 ).wire_format() 

612 path = f"/api/v1/game/first_game/player/{new_player_id}" 

613 req, resp = reusable_client.put(path, data=data) 

614 assert resp.status == 204 

615 player = await queries.select_game_player( 

616 conn=db_conn, 

617 game_id="first_game", 

618 player_id=new_player_id, 

619 ) 

620 assert player == models.GamePlayer( 

621 game_id="first_game", 

622 id=new_player_id, 

623 name="Bob", 

624 team=Team.South, 

625 score=6969, 

626 ) 

627 

628 # Delete existing -> NO CONTENT. 

629 path = f"/api/v1/game/first_game/player/{new_player_id}" 

630 req, resp = reusable_client.delete(path) 

631 assert resp.status == 204 

632 

633 # Second delete, should already be gone -> 404. 

634 path = f"/api/v1/game/first_game/player/{new_player_id}" 

635 req, resp = reusable_client.delete(path) 

636 assert resp.status == 404 

637 

638 # Delete player that never existed -> 404. 

639 path = "/api/v1/game/first_game/player/6904234" 

640 req, resp = reusable_client.delete(path) 

641 assert resp.status == 404 

642 player = await queries.select_game_player( 

643 conn=db_conn, 

644 game_id="first_game", 

645 player_id=6904234, 

646 ) 

647 assert player is None 

648 

649 # Put with invalid ID (not int) -> 404 (Sanic logic). 

650 path = "/api/v1/game/first_game/player/asdasd" 

651 req, resp = reusable_client.put(path) 

652 assert resp.status == 404 

653 

654 # Put with invalid data. 

655 data = "asdasdasd\n\n\n\n" 

656 path = "/api/v1/game/first_game/player/0" 

657 req, resp = reusable_client.put(path, data=data) 

658 assert resp.status == 400 

659 

660 

661@pytest.mark.asyncio 

662async def test_api_v1_put_game_objective_state(api_fixture, caplog) -> None: 

663 caplog.set_level(logging.DEBUG) 

664 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

665 

666 # Empty data -> 400. 

667 data = "" 

668 path = "/api/v1/game/first_game/objective_state" 

669 req, resp = reusable_client.put(path, data=data) 

670 assert resp.status == 400 

671 

672 # Empty list (first request) -> clears state -> 201. 

673 data = "[]" 

674 path = "/api/v1/game/first_game/objective_state" 

675 req, resp = reusable_client.put(path, data=data) 

676 assert resp.status == 201 

677 

678 # Empty list (subsequent request) -> clears state -> 204. 

679 data = "[]" 

680 path = "/api/v1/game/first_game/objective_state" 

681 req, resp = reusable_client.put(path, data=data) 

682 assert resp.status == 204 

683 

684 # Valid list (subsequent request) -> 204. 

685 neutral_team = int(Team.Neutral) 

686 data = f"[('BlaBla',0),('SomeObjective',1),('Neutral Objective',{neutral_team})]" 

687 path = "/api/v1/game/first_game/objective_state" 

688 req, resp = reusable_client.put(path, data=data) 

689 assert resp.status == 204 

690 

691 # Valid list (subsequent request) -> 2024. 

692 data = models.GameObjectiveState( 

693 game_id="first_game", 

694 objectives=[ 

695 models.GameObjective( 

696 name="BlaBla", 

697 team_state=models.Team.North, 

698 ), 

699 models.GameObjective( 

700 name="AnotherObjective", 

701 team_state=models.Team.South, 

702 ), 

703 models.GameObjective( 

704 name="THIS IS SOME WEIRD OBJECTIVE 123123", 

705 team_state=models.Team.Neutral, 

706 ), 

707 ] 

708 ).wire_format() 

709 path = "/api/v1/game/first_game/objective_state" 

710 req, resp = reusable_client.put(path, data=data) 

711 assert resp.status == 204 

712 

713 # Bad data -> 400. 

714 data = "asdasd." 

715 path = "/api/v1/game/first_game/objective_state" 

716 req, resp = reusable_client.put(path, data=data) 

717 assert resp.status == 400 

718 

719 # Bad data -> 400. 

720 data = "[(),()]" 

721 path = "/api/v1/game/first_game/objective_state" 

722 req, resp = reusable_client.put(path, data=data) 

723 assert resp.status == 400 

724 

725 # Bad data -> 400. 

726 data = "[" 

727 path = "/api/v1/game/first_game/objective_state" 

728 req, resp = reusable_client.put(path, data=data) 

729 assert resp.status == 400 

730 

731 # Bad data -> 400. 

732 data = "[)" 

733 path = "/api/v1/game/first_game/objective_state" 

734 req, resp = reusable_client.put(path, data=data) 

735 assert resp.status == 400 

736 

737 # Too much data -> 400. 

738 data = "*" * (max_ast_literal_eval_size + 1) 

739 path = "/api/v1/game/first_game/objective_state" 

740 req, resp = reusable_client.put(path, data=data) 

741 assert resp.status == 400 

742 

743 # Valid but wrong Python object. 

744 data = "()" 

745 path = "/api/v1/game/first_game/objective_state" 

746 req, resp = reusable_client.put(path, data=data) 

747 assert resp.status == 400 

748 

749 # Valid but wrong Python object. 

750 data = "[(1,1)]" 

751 path = "/api/v1/game/first_game/objective_state" 

752 req, resp = reusable_client.put(path, data=data) 

753 assert resp.status == 400 

754 

755 # Valid but wrong Python object. 

756 data = "[('ValidString','this_should_be_an_int')]" 

757 path = "/api/v1/game/first_game/objective_state" 

758 req, resp = reusable_client.put(path, data=data) 

759 assert resp.status == 400 

760 

761 # Valid but wrong Python object. 

762 invalid_team_as_int = 696969 

763 data = f"[('ValidString',{invalid_team_as_int})]" 

764 path = "/api/v1/game/first_game/objective_state" 

765 req, resp = reusable_client.put(path, data=data) 

766 assert resp.status == 400 

767 

768 

769@pytest.mark.asyncio 

770async def test_api_v1_post_game_kill(api_fixture, caplog) -> None: 

771 caplog.set_level(logging.DEBUG) 

772 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

773 

774 # Empty data -> 400. 

775 data = "" 

776 path = "/api/v1/game/first_game/kill" 

777 req, resp = reusable_client.post(path, data=data) 

778 assert resp.status == 400 

779 

780 # Bad data -> 400. 

781 data = "\n\n\nasd\n" 

782 path = "/api/v1/game/first_game/kill" 

783 req, resp = reusable_client.post(path, data=data) 

784 assert resp.status == 400 

785 

786 # Valid request. 

787 data = "353.4503560\nSome guy lmao\nI'mDead:(\n0\n1\nRODmgType_SomeTypeLol\n88.53" 

788 path = "/api/v1/game/first_game/kill" 

789 req, resp = reusable_client.post(path, data=data) 

790 assert resp.status == 204 

791 kills = await queries.select_game_kills(conn=db_conn, game_id="first_game") 

792 assert kills 

793 

794 # Valid request (teamkill). 

795 data = "353.4503560\nSome guy lmao\nI'mDead:(\n1\n1\nRODmgType_SomeTypeLol\n88.53" 

796 path = "/api/v1/game/first_game/kill" 

797 req, resp = reusable_client.post(path, data=data) 

798 assert resp.status == 204 

799 kills = await queries.select_game_kills(conn=db_conn, game_id="first_game") 

800 assert kills 

801 

802 

803@pytest.mark.asyncio 

804async def test_api_v1_game_message(api_fixture, caplog) -> None: 

805 caplog.set_level(logging.DEBUG) 

806 api_app, reusable_client, openai_mock_router, steam_mock_router, db_conn = api_fixture 

807 

808 # Non-existent game. 

809 data = "" 

810 path = "/api/v1/game/235235252345234234/message" 

811 req, resp = reusable_client.post(path, data=data) 

812 assert resp.status == 404 

813 

814 # Game was not initialized -> 503 (data does not matter here). 

815 data = "" 

816 path = "/api/v1/game/first_game/message" 

817 req, resp = reusable_client.post(path, data=data) 

818 assert resp.status == 503 

819 

820 # Initialize the game by setting a fake openai_previous_response_id. 

821 # And creating the corresponding query entry. 

822 await db_conn.execute( 

823 """ 

824 UPDATE "game" 

825 SET openai_previous_response_id = 'pytest_dummy_openapi_response_id' 

826 WHERE id = 'first_game'; 

827 """ 

828 ) 

829 

830 # Sneak in a request here -> should be 503 since the query does not exist! 

831 data = "" 

832 path = "/api/v1/game/first_game/message" 

833 req, resp = reusable_client.post(path, data=data) 

834 assert resp.status == 503 

835 

836 await db_conn.execute( 

837 """ 

838 INSERT INTO "openai_query" (time, 

839 game_id, 

840 game_server_address, 

841 game_server_port, 

842 request_length, 

843 response_length, 

844 openai_response_id) 

845 VALUES (NOW() AT TIME ZONE 'UTC', 

846 'first_game', 

847 INET '127.0.0.1', 

848 7777, 

849 69, 

850 6969, 

851 'pytest_dummy_openapi_response_id'); 

852 """ 

853 ) 

854 

855 # Initialized game, bad data -> 400. 

856 data = ""