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

1import os 

2from collections.abc import Iterator 

3from contextlib import asynccontextmanager 

4from typing import Any 

5from unittest.mock import AsyncMock, patch 

6 

7import pytest 

8from fastapi.testclient import TestClient 

9import stamina 

10 

11from app.config import get_settings 

12from app.dependencies.redis_client import get_redis_client 

13from app.main import app 

14 

15TEST_API_KEY = "test-api-key" 

16 

17 

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

19def _use_test_settings() -> None: 

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

21 os.environ["API_KEY"] = TEST_API_KEY 

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

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

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

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

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

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

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

29 get_settings.cache_clear() 

30 

31 

32@pytest.fixture 

33def api_key() -> str: 

34 return TEST_API_KEY 

35 

36 

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

38def mock_s3_client() -> AsyncMock: 

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

40 return AsyncMock() 

41 

42 

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

44def mock_sts_client() -> AsyncMock: 

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

46 return AsyncMock() 

47 

48 

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

50def _patch_aiobotocore( 

51 mock_s3_client: AsyncMock, mock_sts_client: AsyncMock 

52) -> Iterator[None]: 

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

54 

55 @asynccontextmanager 

56 async def _s3_ctx(): 

57 yield mock_s3_client 

58 

59 @asynccontextmanager 

60 async def _sts_ctx(): 

61 yield mock_sts_client 

62 

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

64 if service == "s3": 

65 return _s3_ctx() 

66 if service == "sts": 

67 return _sts_ctx() 

68 raise ValueError( # pragma: no cover 

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

70 ) 

71 

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

73 yield 

74 

75 

76@pytest.fixture(autouse=True) 

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

78 """Clear mock state between tests.""" 

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

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

81 

82 

83@pytest.fixture 

84def mock_redis() -> AsyncMock: 

85 return AsyncMock() 

86 

87 

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

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

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

91 with TestClient(app) as tc: 

92 yield tc 

93 

94 

95@pytest.fixture(name="override_redis", autouse=True) 

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

97 app.dependency_overrides[get_redis_client] = lambda: mock_redis 

98 yield 

99 app.dependency_overrides.clear() 

100 

101 

102@pytest.fixture 

103def disable_redis( 

104 override_redis: Any, monkeypatch: pytest.MonkeyPatch 

105) -> Iterator[None]: 

106 """Disable Redis for a test.""" 

107 old_dep = app.dependency_overrides.get(get_redis_client) 

108 app.dependency_overrides[get_redis_client] = lambda: None 

109 yield 

110 if old_dep is not None: 

111 app.dependency_overrides[get_redis_client] = old_dep 

112 

113 

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

115def set_stamina_testing(): 

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