Coverage for backend / tests / conftest.py: 100%

1import os 

2from collections.abc import Iterator 

3from contextlib import asynccontextmanager 

4from unittest.mock import AsyncMock, patch 

5 

6import pytest 

7from fastapi.testclient import TestClient 

8import stamina 

9 

10from app.config import get_settings 

11from app.dependencies.redis_client import get_redis_client 

12from app.main import app 

13 

14TEST_API_KEY = "test-api-key" 

15 

16 

17@pytest.fixture(scope="module", autouse=True) 

18def _use_test_settings() -> None: 

19 """Env vars must be set before the lifespan runs.""" 

20 os.environ["API_KEY"] = TEST_API_KEY 

21 os.environ["AWS_ACCESS_KEY_ID"] = "fake-access-key" 

22 os.environ["AWS_SECRET_ACCESS_KEY"] = "fake-secret-key" 

23 os.environ["AWS_BUCKET"] = "test-bucket" 

24 os.environ["AWS_REGION"] = "us-east-1" 

25 os.environ["AWS_UPLOAD_ROLE_ARN"] = "arn:aws:iam::123456789012:role/fake" 

26 os.environ["REDIS_URL"] = "redis://localhost:6379/0" 

27 os.environ["GITHUB_TOKEN"] = "fake-github-token" 

28 get_settings.cache_clear() 

29 

30 

31@pytest.fixture 

32def api_key() -> str: 

33 return TEST_API_KEY 

34 

35 

36@pytest.fixture(scope="module") 

37def mock_s3_client() -> AsyncMock: 

38 """Mock for aiobotocore S3 client (AWSStorage._client).""" 

39 return AsyncMock() 

40 

41 

42@pytest.fixture(scope="module") 

43def mock_sts_client() -> AsyncMock: 

44 """Mock for aiobotocore STS client used by AWSStorage.create_upload_session.""" 

45 return AsyncMock() 

46 

47 

48@pytest.fixture(scope="module", autouse=True) 

49def _patch_aiobotocore( 

50 mock_s3_client: AsyncMock, mock_sts_client: AsyncMock 

51) -> Iterator[None]: 

52 """Patch aiobotocore so the lifespan-created AWSStorage gets mocked clients.""" 

53 

54 @asynccontextmanager 

55 async def _s3_ctx(): 

56 yield mock_s3_client 

57 

58 @asynccontextmanager 

59 async def _sts_ctx(): 

60 yield mock_sts_client 

61 

62 def fake_create_client(self, service, **kwargs): 

63 if service == "s3": 

64 return _s3_ctx() 

65 if service == "sts": 

66 return _sts_ctx() 

67 raise ValueError( # pragma: no cover 

68 f"Unexpected AWS service in tests: {service}" 

69 ) 

70 

71 with patch("aiobotocore.session.AioSession.create_client", fake_create_client): 

72 yield 

73 

74 

75@pytest.fixture(autouse=True) 

76def _reset_aws_mocks(mock_s3_client: AsyncMock, mock_sts_client: AsyncMock) -> None: 

77 """Clear mock state between tests.""" 

78 mock_s3_client.reset_mock(return_value=True, side_effect=True) 

79 mock_sts_client.reset_mock(return_value=True, side_effect=True) 

80 

81 

82@pytest.fixture 

83def mock_redis() -> AsyncMock: 

84 return AsyncMock() 

85 

86 

87@pytest.fixture(scope="module") 

88def client(_patch_aiobotocore: None) -> Iterator[TestClient]: 

89 """Lifespan runs once per module; aiobotocore is already patched.""" 

90 with TestClient(app) as tc: 

91 yield tc 

92 

93 

94@pytest.fixture(autouse=True) 

95def _override_redis(mock_redis: AsyncMock) -> Iterator[None]: 

96 app.dependency_overrides[get_redis_client] = lambda: mock_redis 

97 yield 

98 app.dependency_overrides.clear() 

99 

100 

101@pytest.fixture(autouse=True, scope="session") 

102def set_stamina_testing(): 

103 stamina.set_testing(True, attempts=10, cap=True)