Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add type annotations for test_reporting #54

Closed
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion fixity/fixity.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ def scan(
:param str ss_user: Storage service user to authenticate as
:param str ss_key: API key of the storage service user
:param Logger logger: Logger to print output.
:param str report_url: The base URL to a server to which the report will be POSTed after the scan completes. If absent, the report will not be transmitted.
:param str report_url: The base URL to a server to which the report will be POSTed after the scan completes. If absent, the report will not be transmitted.
:param report_auth: Authentication for the report_url. Tupel of (user, password) for HTTP auth.
:param session_id: Identifier for this session, allowing every scan from one run to be identified.
:param bool force_local: If True, will request the Storage Service to perform a local fixity check, instead of using the Space's fixity (if available).
Expand Down
20 changes: 18 additions & 2 deletions fixity/reporting.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,26 @@
import calendar
import json
from datetime import datetime
from typing import Optional
from typing import Tuple

import requests

from .models import Report
from .utils import check_valid_uuid


class ReportServiceException(Exception):
pass


def post_pre_scan_report(aip, start_time, report_url, report_auth=(), session_id=None):
def post_pre_scan_report(
aip: str,
start_time: datetime,
report_url: str,
report_auth: Tuple[Optional[str], Optional[str]] = (),
session_id: str = None,
) -> bool:
"""
Post a pre-scan report to a remote system.

Expand Down Expand Up @@ -51,7 +61,13 @@ def post_pre_scan_report(aip, start_time, report_url, report_auth=(), session_id
return True


def post_success_report(aip, report, report_url, report_auth=(), session_id=None):
def post_success_report(
aip: str,
report: Report,
report_url: str,
report_auth: Tuple[Optional[str], Optional[str]] = (),
session_id: str = None,
) -> Optional[bool]:
"""
POST a JSON fixity scan report to a remote system.

Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,7 @@ ignore_errors = true
[[tool.mypy.overrides]]
module = [
"tests.test_fixity",
"tests.test_reporting",
]
ignore_errors = false

39 changes: 24 additions & 15 deletions tests/test_reporting.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
REPORT_URL = "http://localhost:8003/"


def json_string(filename):
def json_string(filename: str) -> str:
path = os.path.normpath(
os.path.join(__file__, "..", "..", "fixtures", "json", filename)
)
Expand All @@ -24,22 +24,23 @@ def json_string(filename):
@mock.patch(
"requests.post", side_effect=[mock.Mock(status_code=201, spec=requests.Response)]
)
def test_posting_prescan_report(_post):
def test_posting_prescan_report(_post: mock.Mock) -> None:
aip = "be1074fe-217b-46e0-afec-400ea1a2eb36"
start_time = datetime.fromtimestamp(1400022946)
result = reporting.post_pre_scan_report(aip, start_time, REPORT_URL)
assert result is True


def test_posting_prescan_report_raises_on_invalid_uuid():
def test_posting_prescan_report_raises_on_invalid_uuid() -> None:
start_time = datetime.fromtimestamp(1400022946)
with pytest.raises(InvalidUUID):
reporting.post_pre_scan_report("foo", None, None)
reporting.post_pre_scan_report("foo", start_time, REPORT_URL)


@mock.patch(
"requests.post", side_effect=[mock.Mock(status_code=500, spec=requests.Response)]
)
def test_posting_prescan_report_raises_on_500(_post):
def test_posting_prescan_report_raises_on_500(_post: mock.Mock) -> None:
with pytest.raises(reporting.ReportServiceException):
aip = "90cf0850-f5d2-4023-95e2-0e2b7a1e1b8e"
start_time = datetime.fromtimestamp(1400022946)
Expand All @@ -49,14 +50,14 @@ def test_posting_prescan_report_raises_on_500(_post):
@mock.patch(
"requests.post", side_effect=[mock.Mock(status_code=404, spec=requests.Response)]
)
def test_posting_prescan_report_raises_on_404(_post):
def test_posting_prescan_report_raises_on_404(_post: mock.Mock) -> None:
with pytest.raises(reporting.ReportServiceException):
aip = "90cf0850-f5d2-4023-95e2-0e2b7a1e1b8e"
start_time = datetime.fromtimestamp(1400022946)
reporting.post_pre_scan_report(aip, start_time, REPORT_URL)


def test_posting_prescan_report_raises_when_unable_to_connect():
def test_posting_prescan_report_raises_when_unable_to_connect() -> None:
with pytest.raises(reporting.ReportServiceException):
aip = "90cf0850-f5d2-4023-95e2-0e2b7a1e1b8e"
start_time = datetime.fromtimestamp(1400022946)
Expand All @@ -66,7 +67,7 @@ def test_posting_prescan_report_raises_when_unable_to_connect():
@mock.patch(
"requests.post", side_effect=[mock.Mock(status_code=201, spec=requests.Response)]
)
def test_posting_success_report(_post):
def test_posting_success_report(_post: mock.Mock) -> None:
json_report = json_string("test_failed_report.json")
aip = AIP(uuid="ed42aadc-d854-46c6-b455-cd384eef1618")
report = Report(
Expand All @@ -80,15 +81,23 @@ def test_posting_success_report(_post):
reporting.post_success_report(aip.uuid, report, REPORT_URL)


def test_posting_success_report_raises_on_invalid_uuid():
def test_posting_success_report_raises_on_invalid_uuid() -> None:
json_report = json_string("test_failed_report.json")
report = Report(
begun=datetime.fromtimestamp(1400022946),
ended=datetime.fromtimestamp(1400023208),
success=False,
posted=False,
report=json_report,
)
with pytest.raises(InvalidUUID):
reporting.post_success_report("foo", None, None)
reporting.post_success_report("foo", report, REPORT_URL)


@mock.patch(
"requests.post", side_effect=[mock.Mock(status_code=500, spec=requests.Response)]
)
def test_posting_success_report_raises_on_500(_post):
def test_posting_success_report_raises_on_500(_post: mock.Mock) -> None:
with pytest.raises(reporting.ReportServiceException):
json_report = json_string("test_failed_report.json")
aip = AIP(uuid="ed42aadc-d854-46c6-b455-cd384eef1618")
Expand All @@ -106,7 +115,7 @@ def test_posting_success_report_raises_on_500(_post):
@mock.patch(
"requests.post", side_effect=[mock.Mock(status_code=404, spec=requests.Response)]
)
def test_posting_success_report_raises_on_404(_post):
def test_posting_success_report_raises_on_404(_post: mock.Mock) -> None:
with pytest.raises(reporting.ReportServiceException):
json_report = json_string("test_failed_report.json")
aip = AIP(uuid="ed42aadc-d854-46c6-b455-cd384eef1618")
Expand All @@ -121,7 +130,7 @@ def test_posting_success_report_raises_on_404(_post):
reporting.post_success_report(aip.uuid, report, REPORT_URL)


def test_posting_success_report_raises_when_unable_to_connect():
def test_posting_success_report_raises_when_unable_to_connect() -> None:
with pytest.raises(reporting.ReportServiceException):
json_report = json_string("test_failed_report.json")
aip = AIP(uuid="ed42aadc-d854-46c6-b455-cd384eef1618")
Expand All @@ -136,7 +145,7 @@ def test_posting_success_report_raises_when_unable_to_connect():
reporting.post_success_report(aip.uuid, report, "http://foo")


def test_posting_success_report_posted_is_false_on_raise():
def test_posting_success_report_posted_is_false_on_raise() -> None:
try:
json_report = json_string("test_failed_report.json")
aip = AIP(uuid="ed42aadc-d854-46c6-b455-cd384eef1618")
Expand All @@ -155,7 +164,7 @@ def test_posting_success_report_posted_is_false_on_raise():
assert report.posted is False


def test_posting_success_report_success_none():
def test_posting_success_report_success_none() -> None:
json_report = json_string("test_failed_report.json")
aip = AIP(uuid="ed42aadc-d854-46c6-b455-cd384eef1618")
report = Report(
Expand Down