FilesExpand file tree

main

/

conftest.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

144 lines (103 loc) · 3.97 KB

main

/

conftest.py

Copy path

Top

File metadata and controls

• Code
• Blame

144 lines (103 loc) · 3.97 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

"""Shared test fixtures for Bright Data SDK tests."""

import os

import sys

from typing import Any, Dict, Optional

from unittest.mock import AsyncMock, MagicMock

import pytest

# Add src to path

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))

# ---------------------------------------------------------------------------

# Mock HTTP Response

# ---------------------------------------------------------------------------

class MockResponse:

"""Mock aiohttp response with configurable status, JSON, and text."""

def __init__(

self,

status: int = 200,

json_data: Any = None,

text_data: str = "",

headers: Optional[Dict[str, str]] = None,

):

self.status = status

self._json_data = json_data

self._text_data = text_data

self.headers = headers or {}

self.closed = False

async def json(self):

return self._json_data

async def text(self):

return self._text_data

async def release(self):

pass

def close(self):

self.closed = True

class MockContextManager:

"""Mock async context manager wrapping a MockResponse."""

def __init__(self, response: MockResponse):

self._response = response

async def __aenter__(self):

return self._response

async def __aexit__(self, *args):

pass

# ---------------------------------------------------------------------------

# Engine fixtures

# ---------------------------------------------------------------------------

@pytest.fixture

def mock_engine():

"""

Reusable mock AsyncEngine.

The engine's HTTP methods (get, post, get_from_url, post_to_url, etc.)

return MockContextManager by default. Override per-test by setting

the return_value on the method.

Usage:

async with mock_engine.get("/endpoint") as response:

data = await response.json()

"""

engine = MagicMock()

engine.bearer_token = "test_token_123456789"

default_response = MockResponse(200, json_data={})

default_cm = MockContextManager(default_response)

engine.get = MagicMock(return_value=default_cm)

engine.post = MagicMock(return_value=default_cm)

engine.delete = MagicMock(return_value=default_cm)

engine.get_from_url = MagicMock(return_value=default_cm)

engine.post_to_url = MagicMock(return_value=default_cm)

engine.request = MagicMock(return_value=default_cm)

return engine

def make_engine_response(engine, method: str, response: MockResponse):

"""

Configure a mock engine method to return a specific response.

Args:

engine: Mock engine from mock_engine fixture

method: Method name ("get", "post", "get_from_url", "post_to_url")

response: MockResponse to return

Usage:

make_engine_response(mock_engine, "post_to_url", MockResponse(200, {"snapshot_id": "abc"}))

"""

cm = MockContextManager(response)

getattr(engine, method).return_value = cm

# ---------------------------------------------------------------------------

# API Client fixtures

# ---------------------------------------------------------------------------

@pytest.fixture

def mock_api_client():

"""Mock DatasetAPIClient with configurable trigger/status/fetch."""

client = AsyncMock()

client.trigger = AsyncMock(return_value="snap_123")

client.get_status = AsyncMock(return_value="ready")

client.fetch_result = AsyncMock(return_value=[{"title": "Test"}])

return client

# ---------------------------------------------------------------------------

# Helpers

# ---------------------------------------------------------------------------

def make_status_sequence(statuses):

"""

Create an async side_effect that returns statuses in order.

Usage:

mock_api_client.get_status.side_effect = make_status_sequence(

["in_progress", "in_progress", "ready"]

)

"""

async def side_effect(*args, **kwargs):

return statuses.pop(0)

return side_effect