tobias/radicale
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Tests: Simplify HTTP status check

Browse Source
This commit is contained in:
Unrud 2022-01-16 13:07:56 +01:00
parent 9c0b6cdaeb
commit 402bd3580e
4 changed files with 71 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
radicale/tests/__init__.py
View File

@ -65,7 +65,8 @@ class BaseTest:
shutil.rmtree(self.colpath) shutil.rmtree(self.colpath)
def request(self, method: str, path: str, data: Optional[str] = None, def request(self, method: str, path: str, data: Optional[str] = None,
**kwargs) -> Tuple[int, Dict[str, str], str]: check: Optional[int] = None, **kwargs
) -> Tuple[int, Dict[str, str], str]:
"""Send a request.""" """Send a request."""
login = kwargs.pop("login", None) login = kwargs.pop("login", None)
if login is not None and not isinstance(login, str): if login is not None and not isinstance(login, str):
@ -92,13 +93,13 @@ class BaseTest:
def start_response(status_: str, headers_: List[Tuple[str, str]] def start_response(status_: str, headers_: List[Tuple[str, str]]
) -> None: ) -> None:
nonlocal status, headers nonlocal status, headers
status = status_ status = int(status_.split()[0])
headers = headers_ headers = dict(headers_)
answers = list(self.application(environ, start_response)) answers = list(self.application(environ, start_response))
assert status is not None and headers is not None assert status is not None and headers is not None
assert check is None or status == check, "%d != %d" % (status, check)
return (int(status.split()[0]), dict(headers), return status, headers, answers[0].decode() if answers else ""
answers[0].decode() if answers else "")
@staticmethod @staticmethod
def parse_responses(text: str) -> RESPONSES: def parse_responses(text: str) -> RESPONSES:
@ -130,38 +131,30 @@ class BaseTest:
path_responses[href.text] = prop_respones path_responses[href.text] = prop_respones
return path_responses return path_responses
@staticmethod def get(self, path: str, check: Optional[int] = 200, **kwargs
def _check_status(status: int, good_status: int,
check: Union[bool, int] = True) -> bool:
if check is not False:
expected = good_status if check is True else check
assert status == expected, "%d != %d" % (status, expected)
return status == good_status
def get(self, path: str, check: Union[bool, int] = True, **kwargs
) -> Tuple[int, str]: ) -> Tuple[int, str]:
assert "data" not in kwargs assert "data" not in kwargs
status, _, answer = self.request("GET", path, **kwargs) status, _, answer = self.request("GET", path, check=check, **kwargs)
self._check_status(status, 200, check)
return status, answer return status, answer
def post(self, path: str, data: str = None, check: Union[bool, int] = True, def post(self, path: str, data: str = None, check: Optional[int] = 200,
**kwargs) -> Tuple[int, str]: **kwargs) -> Tuple[int, str]:
status, _, answer = self.request("POST", path, data, **kwargs) status, _, answer = self.request("POST", path, data, check=check,
self._check_status(status, 200, check) **kwargs)
return status, answer return status, answer
def put(self, path: str, data: str, check: Union[bool, int] = True, def put(self, path: str, data: str, check: Optional[int] = 201,
**kwargs) -> Tuple[int, str]: **kwargs) -> Tuple[int, str]:
status, _, answer = self.request("PUT", path, data, **kwargs) status, _, answer = self.request("PUT", path, data, check=check,
self._check_status(status, 201, check) **kwargs)
return status, answer return status, answer
def propfind(self, path: str, data: Optional[str] = None, def propfind(self, path: str, data: Optional[str] = None,
check: Union[bool, int] = True, **kwargs check: Optional[int] = 207, **kwargs
) -> Tuple[int, RESPONSES]: ) -> Tuple[int, RESPONSES]:
status, _, answer = self.request("PROPFIND", path, data, **kwargs) status, _, answer = self.request("PROPFIND", path, data, check=check,
if not self._check_status(status, 207, check): **kwargs)
if status < 200 or 300 <= status:
return status, {} return status, {}
assert answer is not None assert answer is not None
responses = self.parse_responses(answer) responses = self.parse_responses(answer)
@ -170,29 +163,31 @@ class BaseTest:
return status, responses return status, responses
def proppatch(self, path: str, data: Optional[str] = None, def proppatch(self, path: str, data: Optional[str] = None,
check: Union[bool, int] = True, **kwargs check: Optional[int] = 207, **kwargs
) -> Tuple[int, RESPONSES]: ) -> Tuple[int, RESPONSES]:
status, _, answer = self.request("PROPPATCH", path, data, **kwargs) status, _, answer = self.request("PROPPATCH", path, data, check=check,
if not self._check_status(status, 207, check): **kwargs)
if status < 200 or 300 <= status:
return status, {} return status, {}
assert answer is not None assert answer is not None
responses = self.parse_responses(answer) responses = self.parse_responses(answer)
assert len(responses) == 1 and path in responses assert len(responses) == 1 and path in responses
return status, responses return status, responses
def report(self, path: str, data: str, check: Union[bool, int] = True, def report(self, path: str, data: str, check: Optional[int] = 207,
**kwargs) -> Tuple[int, RESPONSES]: **kwargs) -> Tuple[int, RESPONSES]:
status, _, answer = self.request("REPORT", path, data, **kwargs) status, _, answer = self.request("REPORT", path, data, check=check,
if not self._check_status(status, 207, check): **kwargs)
if status < 200 or 300 <= status:
return status, {} return status, {}
assert answer is not None assert answer is not None
return status, self.parse_responses(answer) return status, self.parse_responses(answer)
def delete(self, path: str, check: Union[bool, int] = True, **kwargs def delete(self, path: str, check: Optional[int] = 200, **kwargs
) -> Tuple[int, RESPONSES]: ) -> Tuple[int, RESPONSES]:
assert "data" not in kwargs assert "data" not in kwargs
status, _, answer = self.request("DELETE", path, **kwargs) status, _, answer = self.request("DELETE", path, check=check, **kwargs)
if not self._check_status(status, 200, check): if status < 200 or 300 <= status:
return status, {} return status, {}
assert answer is not None assert answer is not None
responses = self.parse_responses(answer) responses = self.parse_responses(answer)
@ -200,19 +195,18 @@ class BaseTest:
return status, responses return status, responses
def mkcalendar(self, path: str, data: Optional[str] = None, def mkcalendar(self, path: str, data: Optional[str] = None,
check: Union[bool, int] = True, **kwargs check: Optional[int] = 201, **kwargs
) -> Tuple[int, str]: ) -> Tuple[int, str]:
status, _, answer = self.request("MKCALENDAR", path, data, **kwargs) status, _, answer = self.request("MKCALENDAR", path, data, check=check,
self._check_status(status, 201, check) **kwargs)
return status, answer return status, answer
def mkcol(self, path: str, data: Optional[str] = None, def mkcol(self, path: str, data: Optional[str] = None,
check: Union[bool, int] = True, **kwargs) -> int: check: Optional[int] = 201, **kwargs) -> int:
status, _, _ = self.request("MKCOL", path, data, **kwargs) status, _, _ = self.request("MKCOL", path, data, check=check, **kwargs)
self._check_status(status, 201, check)
return status return status
def create_addressbook(self, path: str, check: Union[bool, int] = True, def create_addressbook(self, path: str, check: Optional[int] = 201,
**kwargs) -> int: **kwargs) -> int:
assert "data" not in kwargs assert "data" not in kwargs
return self.mkcol(path, """\ return self.mkcol(path, """\

76
radicale/tests/test_base.py
View File

@ -76,8 +76,7 @@ permissions: RrWw""")
event = get_file_content("event1.ics") event = get_file_content("event1.ics")
path = "/calendar.ics/event1.ics" path = "/calendar.ics/event1.ics"
self.put(path, event) self.put(path, event)
status, headers, answer = self.request("GET", path) _, headers, answer = self.request("GET", path, check=200)
assert status == 200
assert "ETag" in headers assert "ETag" in headers
assert headers["Content-Type"] == "text/calendar; charset=utf-8" assert headers["Content-Type"] == "text/calendar; charset=utf-8"
assert "VEVENT" in answer assert "VEVENT" in answer
@ -98,7 +97,7 @@ permissions: RrWw""")
event = get_file_content("event1.ics") event = get_file_content("event1.ics")
self.put("/calendar.ics/event1.ics", event) self.put("/calendar.ics/event1.ics", event)
status, answer = self.put( status, answer = self.put(
"/calendar.ics/event1-duplicate.ics", event, check=False) "/calendar.ics/event1-duplicate.ics", event, check=None)
assert status in (403, 409) assert status in (403, 409)
xml = DefusedET.fromstring(answer) xml = DefusedET.fromstring(answer)
assert xml.tag == xmlutils.make_clark("D:error") assert xml.tag == xmlutils.make_clark("D:error")
@ -116,8 +115,7 @@ permissions: RrWw""")
todo = get_file_content("todo1.ics") todo = get_file_content("todo1.ics")
path = "/calendar.ics/todo1.ics" path = "/calendar.ics/todo1.ics"
self.put(path, todo) self.put(path, todo)
status, headers, answer = self.request("GET", path) _, headers, answer = self.request("GET", path, check=200)
assert status == 200
assert "ETag" in headers assert "ETag" in headers
assert headers["Content-Type"] == "text/calendar; charset=utf-8" assert headers["Content-Type"] == "text/calendar; charset=utf-8"
assert "VTODO" in answer assert "VTODO" in answer
@ -130,8 +128,7 @@ permissions: RrWw""")
contact = get_file_content("contact1.vcf") contact = get_file_content("contact1.vcf")
path = "/contacts.vcf/contact.vcf" path = "/contacts.vcf/contact.vcf"
self.put(path, contact) self.put(path, contact)
status, headers, answer = self.request("GET", path) _, headers, answer = self.request("GET", path, check=200)
assert status == 200
assert "ETag" in headers assert "ETag" in headers
assert headers["Content-Type"] == "text/vcard; charset=utf-8" assert headers["Content-Type"] == "text/vcard; charset=utf-8"
assert "VCARD" in answer assert "VCARD" in answer
@ -174,7 +171,7 @@ permissions: RrWw""")
event2 = get_file_content("event2.ics") event2 = get_file_content("event2.ics")
path = "/calendar.ics/event1.ics" path = "/calendar.ics/event1.ics"
self.put(path, event1) self.put(path, event1)
status, answer = self.put(path, event2, check=False) status, answer = self.put(path, event2, check=None)
assert status in (403, 409) assert status in (403, 409)
xml = DefusedET.fromstring(answer) xml = DefusedET.fromstring(answer)
assert xml.tag == xmlutils.make_clark("D:error") assert xml.tag == xmlutils.make_clark("D:error")
@ -267,7 +264,7 @@ permissions: RrWw""")
def test_mkcalendar_overwrite(self) -> None: def test_mkcalendar_overwrite(self) -> None:
"""Try to overwrite an existing calendar.""" """Try to overwrite an existing calendar."""
self.mkcalendar("/calendar.ics/") self.mkcalendar("/calendar.ics/")
status, answer = self.mkcalendar("/calendar.ics/", check=False) status, answer = self.mkcalendar("/calendar.ics/", check=None)
assert status in (403, 409) assert status in (403, 409)
xml = DefusedET.fromstring(answer) xml = DefusedET.fromstring(answer)
assert xml.tag == xmlutils.make_clark("D:error") assert xml.tag == xmlutils.make_clark("D:error")
@ -276,8 +273,7 @@ permissions: RrWw""")
def test_mkcalendar_intermediate(self) -> None: def test_mkcalendar_intermediate(self) -> None:
"""Try make a calendar in a unmapped collection.""" """Try make a calendar in a unmapped collection."""
status, _ = self.mkcalendar("/unmapped/calendar.ics/", check=False) self.mkcalendar("/unmapped/calendar.ics/", check=409)
assert status == 409
def test_mkcol(self) -> None: def test_mkcol(self) -> None:
"""Make a collection.""" """Make a collection."""
@ -286,13 +282,11 @@ permissions: RrWw""")
def test_mkcol_overwrite(self) -> None: def test_mkcol_overwrite(self) -> None:
"""Try to overwrite an existing collection.""" """Try to overwrite an existing collection."""
self.mkcol("/user/") self.mkcol("/user/")
status = self.mkcol("/user/", check=False) self.mkcol("/user/", check=405)
assert status == 405
def test_mkcol_intermediate(self) -> None: def test_mkcol_intermediate(self) -> None:
"""Try make a collection in a unmapped collection.""" """Try make a collection in a unmapped collection."""
status = self.mkcol("/unmapped/user/", check=False) self.mkcol("/unmapped/user/", check=409)
assert status == 409
def test_mkcol_make_calendar(self) -> None: def test_mkcol_make_calendar(self) -> None:
"""Make a calendar with additional props.""" """Make a calendar with additional props."""
@ -317,9 +311,8 @@ permissions: RrWw""")
path1 = "/calendar.ics/event1.ics" path1 = "/calendar.ics/event1.ics"
path2 = "/calendar.ics/event2.ics" path2 = "/calendar.ics/event2.ics"
self.put(path1, event) self.put(path1, event)
status, _, _ = self.request( self.request("MOVE", path1, check=201,
"MOVE", path1, HTTP_DESTINATION=path2, HTTP_HOST="") HTTP_DESTINATION=path2, HTTP_HOST="")
assert status == 201
self.get(path1, check=404) self.get(path1, check=404)
self.get(path2) self.get(path2)
@ -331,9 +324,8 @@ permissions: RrWw""")
path1 = "/calendar1.ics/event1.ics" path1 = "/calendar1.ics/event1.ics"
path2 = "/calendar2.ics/event2.ics" path2 = "/calendar2.ics/event2.ics"
self.put(path1, event) self.put(path1, event)
status, _, _ = self.request( self.request("MOVE", path1, check=201,
"MOVE", path1, HTTP_DESTINATION=path2, HTTP_HOST="") HTTP_DESTINATION=path2, HTTP_HOST="")
assert status == 201
self.get(path1, check=404) self.get(path1, check=404)
self.get(path2) self.get(path2)
@ -362,12 +354,10 @@ permissions: RrWw""")
path2 = "/calendar2.ics/event1.ics" path2 = "/calendar2.ics/event1.ics"
self.put(path1, event) self.put(path1, event)
self.put(path2, event) self.put(path2, event)
status, _, _ = self.request( self.request("MOVE", path1, check=412,
"MOVE", path1, HTTP_DESTINATION=path2, HTTP_HOST="") HTTP_DESTINATION=path2, HTTP_HOST="")
assert status == 412 self.request("MOVE", path1, check=204,
status, _, _ = self.request("MOVE", path1, HTTP_DESTINATION=path2, HTTP_DESTINATION=path2, HTTP_HOST="", HTTP_OVERWRITE="T")
HTTP_HOST="", HTTP_OVERWRITE="T")
assert status == 204
def test_move_between_colections_overwrite_uid_conflict(self) -> None: def test_move_between_colections_overwrite_uid_conflict(self) -> None:
"""Move a item to a collection which already contains the item with """Move a item to a collection which already contains the item with
@ -388,13 +378,11 @@ permissions: RrWw""")
assert xml.find(xmlutils.make_clark("C:no-uid-conflict")) is not None assert xml.find(xmlutils.make_clark("C:no-uid-conflict")) is not None
def test_head(self) -> None: def test_head(self) -> None:
status, headers, answer = self.request("HEAD", "/") _, headers, answer = self.request("HEAD", "/", check=302)
assert status == 302
assert int(headers.get("Content-Length", "0")) > 0 and not answer assert int(headers.get("Content-Length", "0")) > 0 and not answer
def test_options(self) -> None: def test_options(self) -> None:
status, headers, _ = self.request("OPTIONS", "/") _, headers, _ = self.request("OPTIONS", "/", check=200)
assert status == 200
assert "DAV" in headers assert "DAV" in headers
def test_delete_collection(self) -> None: def test_delete_collection(self) -> None:
@ -650,7 +638,7 @@ permissions: RrWw""")
report = "addressbook-query" report = "addressbook-query"
else: else:
raise ValueError("Unsupported kind: %r" % kind) raise ValueError("Unsupported kind: %r" % kind)
status, _, = self.delete(path, check=False) status, _, = self.delete(path, check=None)
assert status in (200, 404) assert status in (200, 404)
create_collection_fn(path) create_collection_fn(path)
for i in items: for i in items:
@ -1439,9 +1427,8 @@ permissions: RrWw""")
self.put(event1_path, event) self.put(event1_path, event)
sync_token, responses = self._report_sync_token(calendar_path) sync_token, responses = self._report_sync_token(calendar_path)
assert len(responses) == 1 and responses[event1_path] == 200 assert len(responses) == 1 and responses[event1_path] == 200
status, _, _ = self.request( self.request("MOVE", event1_path, check=201,
"MOVE", event1_path, HTTP_DESTINATION=event2_path, HTTP_HOST="") HTTP_DESTINATION=event2_path, HTTP_HOST="")
assert status == 201
sync_token, responses = self._report_sync_token( sync_token, responses = self._report_sync_token(
calendar_path, sync_token) calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token: if not self.full_sync_token_support and not sync_token:
@ -1459,12 +1446,10 @@ permissions: RrWw""")
self.put(event1_path, event) self.put(event1_path, event)
sync_token, responses = self._report_sync_token(calendar_path) sync_token, responses = self._report_sync_token(calendar_path)
assert len(responses) == 1 and responses[event1_path] == 200 assert len(responses) == 1 and responses[event1_path] == 200
status, _, _ = self.request( self.request("MOVE", event1_path, check=201,
"MOVE", event1_path, HTTP_DESTINATION=event2_path, HTTP_HOST="") HTTP_DESTINATION=event2_path, HTTP_HOST="")
assert status == 201 self.request("MOVE", event2_path, check=201,
status, _, _ = self.request( HTTP_DESTINATION=event1_path, HTTP_HOST="")
"MOVE", event2_path, HTTP_DESTINATION=event1_path, HTTP_HOST="")
assert status == 201
sync_token, responses = self._report_sync_token( sync_token, responses = self._report_sync_token(
calendar_path, sync_token) calendar_path, sync_token)
if not self.full_sync_token_support and not sync_token: if not self.full_sync_token_support and not sync_token:
@ -1518,7 +1503,7 @@ permissions: RrWw""")
self.mkcalendar("/test/") self.mkcalendar("/test/")
for component in ("event", "todo", "journal"): for component in ("event", "todo", "journal"):
event = get_file_content("%s1.ics" % component) event = get_file_content("%s1.ics" % component)
status, _ = self.delete("/test/test.ics", check=False) status, _ = self.delete("/test/test.ics", check=None)
assert status in (200, 404) assert status in (200, 404)
self.put("/test/test.ics", event) self.put("/test/test.ics", event)
_, responses = self.report("/test/", """\ _, responses = self.report("/test/", """\
@ -1607,12 +1592,11 @@ permissions: RrWw""")
def test_custom_headers(self) -> None: def test_custom_headers(self) -> None:
self.configure({"headers": {"test": "123"}}) self.configure({"headers": {"test": "123"}})
# Test if header is set on success # Test if header is set on success
status, headers, _ = self.request("OPTIONS", "/") _, headers, _ = self.request("OPTIONS", "/", check=200)
assert status == 200
assert headers.get("test") == "123" assert headers.get("test") == "123"
# Test if header is set on failure # Test if header is set on failure
status, headers, _ = self.request("GET", "/.well-known/does not exist") _, headers, _ = self.request("GET", "/.well-known/does not exist",
assert status == 404 check=404)
assert headers.get("test") == "123" assert headers.get("test") == "123"
@pytest.mark.skipif(sys.version_info < (3, 6), @pytest.mark.skipif(sys.version_info < (3, 6),

8
radicale/tests/test_server.py
View File

@ -97,7 +97,8 @@ class TestBaseServerRequests(BaseTest):
super().teardown() super().teardown()
def request(self, method: str, path: str, data: Optional[str] = None, def request(self, method: str, path: str, data: Optional[str] = None,
**kwargs) -> Tuple[int, Dict[str, str], str]: check: Optional[int] = None, **kwargs
) -> Tuple[int, Dict[str, str], str]:
"""Send a request.""" """Send a request."""
login = kwargs.pop("login", None) login = kwargs.pop("login", None)
if login is not None and not isinstance(login, str): if login is not None and not isinstance(login, str):
@ -128,6 +129,8 @@ class TestBaseServerRequests(BaseTest):
with self.opener.open(req) as f: with self.opener.open(req) as f:
return f.getcode(), dict(f.info()), f.read().decode() return f.getcode(), dict(f.info()), f.read().decode()
except HTTPError as e: except HTTPError as e:
assert check is None or e.code == check, "%d != %d" % (e.code,
check)
return e.code, dict(e.headers), e.read().decode() return e.code, dict(e.headers), e.read().decode()
except URLError as e: except URLError as e:
if not isinstance(e.reason, ConnectionRefusedError): if not isinstance(e.reason, ConnectionRefusedError):
@ -209,8 +212,7 @@ class TestBaseServerRequests(BaseTest):
env={**os.environ, "PYTHONPATH": os.pathsep.join(sys.path)}) env={**os.environ, "PYTHONPATH": os.pathsep.join(sys.path)})
try: try:
status, headers, _ = self.request( status, headers, _ = self.request(
"GET", "/", is_alive_fn=lambda: p.poll() is None) "GET", "/", check=302, is_alive_fn=lambda: p.poll() is None)
self._check_status(status, 302)
for key in self.configuration.options("headers"): for key in self.configuration.options("headers"):
assert headers.get(key) == self.configuration.get( assert headers.get(key) == self.configuration.get(
"headers", key) "headers", key)

3
radicale/tests/test_web.py
View File

@ -26,8 +26,7 @@ class TestBaseWebRequests(BaseTest):
"""Test web plugin.""" """Test web plugin."""
def test_internal(self) -> None: def test_internal(self) -> None:
status, headers, _ = self.request("GET", "/.web") _, headers, _ = self.request("GET", "/.web", check=302)
assert status == 302
assert headers.get("Location") == ".web/" assert headers.get("Location") == ".web/"
_, answer = self.get("/.web/") _, answer = self.get("/.web/")
assert answer assert answer