forked from ChelseaKR/mrf-honest
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_probe.py
More file actions
180 lines (144 loc) · 5.46 KB
/
Copy pathtest_probe.py
File metadata and controls
180 lines (144 loc) · 5.46 KB
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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
"""Behavioral tests for the bounded format probe."""
from __future__ import annotations
from urllib.error import HTTPError
from urllib.request import Request
import pytest
import robots_fixtures
from mrf_honest.fetch import FetchPolicy, ProbeOutcome, probe_url
from mrf_honest.types import ResponseLike
URL = "https://files.example.test/standardcharges"
CSV_HEAD = (
b"\xef\xbb\xbfhospital_name,last_updated_on,version,location_name,hospital_address,"
b'license_number|CA,type_2_npi,"attestation...",attester_name\r\n'
)
class _ProbeResponse:
"""A ResponseLike that can honor or ignore a Range request."""
def __init__(
self,
body: bytes,
*,
url: str = URL,
status: int = 206,
headers: dict[str, str] | None = None,
) -> None:
self._body = body
self.status = status
self.headers = headers if headers is not None else {}
self._url = url
self.reads: list[int] = []
self.closed = False
def read(self, amount: int = -1) -> bytes:
self.reads.append(amount)
if amount < 0:
return self._body
return self._body[:amount]
def geturl(self) -> str:
return self._url
def close(self) -> None:
self.closed = True
class _Opener:
def __init__(self, response: ResponseLike | Exception) -> None:
self._response = response
self.requests: list[Request] = []
def __call__(self, request: Request, *, timeout: float) -> ResponseLike:
self.requests.append(request)
if isinstance(self._response, Exception):
raise self._response
return self._response
def _policy() -> FetchPolicy:
return FetchPolicy(contact="ops@example.test")
def _probe(response: ResponseLike | Exception, **kwargs: object) -> ProbeOutcome:
opener = _Opener(response)
outcome = probe_url(
URL,
policy=_policy(),
politeness=robots_fixtures.politeness(),
opener=opener,
**kwargs, # type: ignore[arg-type]
)
return outcome
def test_a_ranged_sample_is_requested_and_classified() -> None:
response = _ProbeResponse(
CSV_HEAD,
status=206,
headers={"Content-Type": "text/csv", "Content-Range": "bytes 0-4095/34786529"},
)
opener = _Opener(response)
outcome = probe_url(
URL, policy=_policy(), politeness=robots_fixtures.politeness(), opener=opener
)
assert outcome.status == "probed"
assert outcome.range_honored is True
assert outcome.sniffed == "text"
assert outcome.starts_with_csv_general_header is True
assert outcome.declared_size == 34786529
assert outcome.content_type == "text/csv"
request = opener.requests[0]
assert request.get_header("Range") == "bytes=0-4095"
assert request.get_header("Accept-encoding") == "identity"
assert response.closed is True
def test_a_server_that_ignores_range_is_read_up_to_the_bound_only() -> None:
body = b"{" + b" " * 100_000
response = _ProbeResponse(body, status=200, headers={"Content-Length": str(len(body))})
outcome = _probe(response, sample_bytes=1_024)
assert outcome.status == "probed"
assert outcome.range_honored is False
assert outcome.bytes_sampled == 1_024
assert outcome.sniffed == "json"
assert outcome.declared_size == len(body)
assert response.reads == [1_024]
@pytest.mark.parametrize(
("body", "expected"),
[
(b"PK\x03\x04rest-of-zip", "zip"),
(b"\x1f\x8b\x08gzip-members", "gzip"),
(b' {"hospital_name": "x"}', "json"),
(b"<!DOCTYPE html><html>", "html"),
(b"plain,comma,text\r\n", "text"),
(b"\x00\x01\x02\x03", "binary"),
(b"", "empty"),
],
)
def test_leading_bytes_decide_the_classification(body: bytes, expected: str) -> None:
response = _ProbeResponse(body, status=206)
outcome = _probe(response)
assert outcome.sniffed == expected
def test_a_bom_is_stripped_before_classification() -> None:
outcome = _probe(_ProbeResponse(b"\xef\xbb\xbf[1]", status=206))
assert outcome.sniffed == "json"
def test_robots_disallow_stops_the_probe_before_any_request() -> None:
opener = _Opener(_ProbeResponse(b"{}"))
outcome = probe_url(
URL,
policy=_policy(),
politeness=robots_fixtures.politeness("User-agent: *\nDisallow: /\n"),
opener=opener,
)
assert outcome.status == "robots_disallowed"
assert opener.requests == []
def test_an_http_error_is_recorded_not_raised() -> None:
import email.message
import io
headers = email.message.Message()
headers["Content-Type"] = "text/html"
error = HTTPError(URL, 403, "Forbidden", headers, io.BytesIO(b""))
outcome = _probe(error)
assert outcome.status == "http_error"
assert outcome.http_status == 403
assert outcome.content_type == "text/html"
def test_an_invalid_url_never_reaches_the_network() -> None:
opener = _Opener(_ProbeResponse(b"{}"))
outcome = probe_url(
"http://insecure.example.test/prices.csv",
policy=_policy(),
politeness=robots_fixtures.politeness(),
opener=opener,
)
assert outcome.status == "invalid_url"
assert opener.requests == []
def test_the_outcome_serializes_to_json_safe_values() -> None:
outcome = _probe(_ProbeResponse(CSV_HEAD, status=206))
payload = outcome.to_dict()
assert payload["status"] == "probed"
assert isinstance(payload["sample_sha256"], str)
assert payload["starts_with_csv_general_header"] is True