86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
import asyncio
|
|
import types
|
|
import sys
|
|
from pathlib import Path
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_lot_bidding_data_403(monkeypatch):
|
|
"""
|
|
Simulate a 403 from the GraphQL endpoint and verify:
|
|
- Function returns None (graceful handling)
|
|
- It attempts a retry and logs a clear 403 message
|
|
"""
|
|
# Load modules directly from src using importlib to avoid path issues
|
|
project_root = Path(__file__).resolve().parents[1]
|
|
src_path = project_root / 'src'
|
|
import importlib.util
|
|
|
|
def _load_module(name, file_path):
|
|
spec = importlib.util.spec_from_file_location(name, str(file_path))
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module) # type: ignore
|
|
return module
|
|
|
|
# Load config first because graphql_client imports it by module name
|
|
config = _load_module('config', src_path / 'config.py')
|
|
graphql_client = _load_module('graphql_client', src_path / 'graphql_client.py')
|
|
monkeypatch.setattr(config, "OFFLINE", False, raising=False)
|
|
|
|
log_messages = []
|
|
|
|
def fake_print(*args, **kwargs):
|
|
msg = " ".join(str(a) for a in args)
|
|
log_messages.append(msg)
|
|
|
|
import builtins
|
|
monkeypatch.setattr(builtins, "print", fake_print)
|
|
|
|
class MockResponse:
|
|
def __init__(self, status=403, text_body="Forbidden"):
|
|
self.status = status
|
|
self._text_body = text_body
|
|
|
|
async def json(self):
|
|
return {}
|
|
|
|
async def text(self):
|
|
return self._text_body
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
class MockSession:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def post(self, *args, **kwargs):
|
|
# Always return 403
|
|
return MockResponse(403, "Forbidden by WAF")
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
# Patch aiohttp.ClientSession to our mock
|
|
import types as _types
|
|
dummy_aiohttp = _types.SimpleNamespace()
|
|
dummy_aiohttp.ClientSession = MockSession
|
|
# Ensure that an `import aiohttp` inside the function resolves to our dummy
|
|
monkeypatch.setitem(sys.modules, 'aiohttp', dummy_aiohttp)
|
|
|
|
result = await graphql_client.fetch_lot_bidding_data("A1-40179-35")
|
|
|
|
# Should gracefully return None
|
|
assert result is None
|
|
|
|
# Should have logged a 403 at least once
|
|
assert any("GraphQL API error: 403" in m for m in log_messages)
|