Testing your automation with pytest
Fixtures, tmp_path, monkeypatch, stub commands, a local HTTP server and exit codes.
tar -xzf ps-testing.tar.gz, which creates ps-testing/. SHA-256: 02fc63eafd14f9b6260408822272e07e39635dc8f2333471e11692d41b6b2994Automation gets changed long after it was written: a new IOC format, a different API, a quick fix at 2 a.m. Tests are how you change it without breaking what already worked. In this lesson you will test two tools from earlier lessons with pytest: hashcheck from the argparse lesson and the feed client from the errors lesson. You will write plain assert tests, run one case per input with parametrize, read a failing test line by line, use fixtures for temporary files, environment changes and captured output, replace a function the right way (and see the wrong way fail), put a stub program on PATH that fails or hangs, and run a real HTTP server inside the tests.
pytest in a venv, and the first tests
pytest is a third-party package, so it goes into the project's venv, pinned. Unpack the lesson files (the box at the top of this page) in your home directory and work in ~/ps-testing. They hold the code under test, unchanged copies of the hashcheck modules from the argparse lesson and of feed.py, retry.py and check_feed.py from the errors lesson, the test files shown below, and two small files for the tests:
# Test tools, exact versions: pytest and the packages it needs (the packaging lesson adds hashes).pytest==9.1.1pluggy==1.6.0iniconfig==2.3.0packaging==26.3pygments==2.21.0
[pytest]testpaths = ["tests"] # what a bare `pytest` collectspythonpath = ["."] # tests import the modules in this directory
pytest.toml is pytest's own configuration file (new in pytest 9.0; [tool.pytest] in pyproject.toml works too). testpaths says where a bare pytest looks for tests, and pythonpath puts the project directory on sys.path so tests can import hashing. The packaging lesson replaces that line with an installed package.
pytest collects files named test_*.py and runs every function in them whose name starts with test_. A test is a function that calls your code and checks the result with a plain assert. The first file tests load_iocs():
"""load_iocs(): which IOC lists are accepted, and what they turn into."""import ioimport pytestfrom hashing import IocListError, load_iocsDIGEST = "ab" * 32 # 64 hex digits: the shape of a SHA-256def test_indicator_with_label():iocs = load_iocs(io.StringIO(f"{DIGEST} dropper-2026-091\n"), "test")assert iocs == {DIGEST: "dropper-2026-091"}@pytest.mark.parametrize("line, expected",[(DIGEST, {DIGEST: "unlabelled"}),(f"{DIGEST.upper()} loud", {DIGEST: "loud"}),(f" {DIGEST} spaced # a comment", {DIGEST: "spaced"}),],ids=["no-label", "upper-case", "comment"],)def test_accepted_line(line, expected):assert load_iocs(io.StringIO(line + "\n"), "test") == expected@pytest.mark.parametrize("text",["", "# only a comment\n", "e3b0c44298fc1c14 short-hash\n", DIGEST + "0\n"],ids=["empty", "comment-only", "short", "65-digits"],)def test_rejected_list(text):with pytest.raises(IocListError):load_iocs(io.StringIO(text), "test")
io.StringIO is a string that behaves like an open text file, so no file is needed. A line starting with @ is a decorator, as the functions lesson showed: it hands the function defined below it to another function, here one that registers it with pytest. @pytest.mark.parametrize runs the test once per row, with the row's values as arguments; ids names each case. with pytest.raises(IocListError): passes only if the block raises that exception, which is how a test checks a failure path.
Eight tests from three functions: each parametrized row is a separate test with its own ID in brackets. The header names the Python and pytest versions, the root directory and the configuration file that was used. Each dot in the first run is a passed test.
Reading a failing test
Now a realistic mistake: an edit removes the .lower() that normalises the hash, and sed makes that change in the lab:
Read it from the top. ..F..... shows the third test failed. Under FAILURES, the heading names it, test_accepted_line[upper-case], and the next lines show the parameter values it received. The source follows, and > marks the line that failed. The E lines explain: the returned dictionary has an upper-case key where the test expected a lower-case one. tests/test_hashing.py:26 is the file and line. short test summary info lists every failure in one line each, and pytest's exit status is 1, which is what a CI job checks (0 means all passed, 5 that no tests were collected). The bug matters: an IOC list published in upper case would stop matching anything, and every scan would report "no match". Put the line back:
Fixtures: tmp_path, monkeypatch, capsys and caplog
A fixture is a function that prepares something a test needs. A test asks for a fixture by naming it as a parameter; pytest calls the fixture, passes in its value, and cleans up after the test. Fixtures in conftest.py are available to every test file in its directory, without an import:
"""Fixtures shared by the test files in this directory (pytest loads conftest.py itself)."""import hashlibimport pytestSAMPLE_A = b"lab sample A: stands in for a malicious file\n"SAMPLE_B = b"lab sample B: a different file with the same name\n"@pytest.fixturedef workdir(tmp_path, monkeypatch):"""A small upload tree and IOC list in a fresh temporary directory; the test runs inside it."""(tmp_path / "uploads" / "a").mkdir(parents=True)(tmp_path / "uploads" / "b").mkdir()(tmp_path / "uploads" / "a" / "invoice.pdf").write_bytes(SAMPLE_A)(tmp_path / "uploads" / "b" / "invoice.pdf").write_bytes(SAMPLE_B)(tmp_path / "uploads" / "b" / "notes.txt").write_text("nothing to see\n")(tmp_path / "iocs.txt").write_text(f"{hashlib.sha256(SAMPLE_A).hexdigest()} sample-a\n"f"{hashlib.sha256(SAMPLE_B).hexdigest()} sample-b\n")monkeypatch.chdir(tmp_path) # undone after the test, like every monkeypatch changereturn tmp_path
workdir uses two of pytest's built-in fixtures. tmp_path is a new, empty directory for each test (a pathlib.Path), so tests never share files or touch the project; pytest keeps the last three runs under /tmp/pytest-of-USER for inspection. monkeypatch changes something for one test and undoes it afterwards: here the working directory, later attributes and environment variables. The tests drive hashcheck through main(argv), the way the argparse lesson designed it:
"""hashcheck through main(argv): exit status, stdout, log messages and the files on disk."""import pytestimport hashcheckfrom hashcheck import mainA, B = "uploads/a/invoice.pdf", "uploads/b/invoice.pdf"def files_under(root):return sorted(str(p.relative_to(root)) for p in root.rglob("*") if p.is_file())@pytest.mark.parametrize("argv, status",[(["scan", "--iocs", "iocs.txt", "uploads/b/notes.txt"], 0),(["scan", "--iocs", "iocs.txt", A, "uploads/b/notes.txt"], 1),(["scan", "--iocs", "missing.txt", A], 2),(["scan", "--iocs", "iocs.txt", A, "nope.pdf"], 3),],ids=["clean", "match", "no-ioc-list", "incomplete"],)def test_exit_status(workdir, argv, status):assert main(argv) == statusdef test_usage_error_exits_2(workdir, capsys):with pytest.raises(SystemExit) as exc: # argparse ends the program with SystemExit(2)main(["scan", "--iocs", "iocs.txt", "--format", "yaml", A])assert exc.value.code == 2assert "invalid choice: 'yaml'" in capsys.readouterr().errdef test_data_on_stdout_problems_in_the_log(workdir, capsys, caplog):assert main(["scan", "--iocs", "iocs.txt", A, "nope.pdf"]) == 3assert capsys.readouterr().out == f"{A} sample-a\n"assert caplog.messages == ["nope.pdf: No such file or directory"]def test_dry_run_changes_nothing(workdir):before = files_under(workdir)assert main(["quarantine", "--iocs", "iocs.txt", "--into", "q", "--dry-run", A, B]) == 1assert files_under(workdir) == beforedef test_same_name_files_are_both_kept(workdir):assert main(["quarantine", "--iocs", "iocs.txt", "--into", "q", A, B]) == 1stored = [name for name in files_under(workdir / "q") if name.endswith("-invoice.pdf")]assert len(stored) == 2assert (workdir / "q").stat().st_mode & 0o777 == 0o700def test_internal_error_is_not_a_finding(workdir, monkeypatch, caplog):def broken(path, max_size):raise RuntimeError("a bug")monkeypatch.setattr(hashcheck, "digest", broken) # the name check() looks upassert main(["scan", "--iocs", "iocs.txt", A]) == 70assert "internal error" in caplog.text
test_exit_status checks the exit-code contract row by row. main() returns most statuses, but argparse ends a usage error with SystemExit(2), so that test catches the exception and reads its code. capsys captures what the code writes to stdout and stderr; readouterr() returns both. caplog captures log records; caplog.messages is their text. Together they prove that the match went to stdout and the warning only to the log. Under pytest, logging.basicConfig() in main() does nothing, because pytest has already attached its capture handler to the root logger and basicConfig() only configures a logger without handlers. Two tests check effects on disk, not messages: the file list is identical after a dry run, and two same-named samples are both kept, in a 0700 directory. The last one forces the internal-error path of the contract: monkeypatch.setattr() replaces hashcheck.digest with a function that raises, as the argparse lesson did by hand, and the test checks for 70, not 1. Why the patch goes on hashcheck and not on hashing is the next section's subject.
pytest --fixtures lists every fixture available to a test file with its docstring, including your own.
monkeypatch: patch the name the code looks up
check_feed.py from the errors lesson calls a feed over HTTP and retries temporary failures. To test its logic without a network, replace get_json with a fake. check_feed.py starts with from feed import ... get_json, and that detail decides where the fake must go. This test is wrong on purpose:
"""WRONG on purpose: patches feed.get_json, but check_feed calls the name it imported."""import check_feedimport feeddef fake_get_json(url, timeout):return {"indicator": "203.0.113.42", "verdict": "malicious"}def test_verdict_printed(monkeypatch):monkeypatch.setattr(feed, "get_json", fake_get_json) # check_feed.get_json is untouchedassert check_feed.main(["http://127.0.0.1:9/indicator"]) == 0
from feed import get_json copies a reference into check_feed. main() looks up check_feed.get_json, so only a patch on that name takes effect.The test tried to reach 127.0.0.1 port 9 for real, was refused four times with real, randomly jittered waits, and main() returned 3. pytest printed the log records captured during the call, which show exactly that. Against a real URL, the same mistake would query the production feed from every test run. The fix is to patch the name in the module that uses it:
"""check_feed.main() with the network replaced: patch the names the code looks up."""import jsonimport timeimport check_feedfrom feed import PermanentFeedError, TemporaryFeedErrorVERDICT = {"indicator": "203.0.113.42", "verdict": "malicious"}def test_verdict_on_stdout(monkeypatch, capsys):def fake_get_json(url, timeout):return VERDICTmonkeypatch.setattr(check_feed, "get_json", fake_get_json) # the name check_feed callsassert check_feed.main(["http://feed.test/indicator"]) == 0assert json.loads(capsys.readouterr().out) == VERDICTdef test_permanent_error_is_not_retried(monkeypatch, caplog):calls = []def not_found(url, timeout):calls.append(url)raise PermanentFeedError("HTTP 404 from the test")monkeypatch.setattr(check_feed, "get_json", not_found)assert check_feed.main(["http://feed.test/nothing"]) == 1assert len(calls) == 1assert "not retried" in caplog.textdef test_gives_up_without_a_final_sleep(monkeypatch, caplog):def down(url, timeout):raise TemporaryFeedError("HTTP 503 from the test")sleeps = []monkeypatch.setattr(check_feed, "get_json", down)# retry.py calls time.sleep(...): it looks sleep up on the time module at every call.monkeypatch.setattr(time, "sleep", sleeps.append)assert check_feed.main(["http://feed.test/down"]) == 3assert len(sleeps) == 3 # four attempts, three waits, none after the lastassert [r.levelname for r in caplog.records].count("WARNING") == 3
The last test also replaces time.sleep, and there patching the time module is right: retry.py does import time and looks up time.sleep on the module at every call. Replacing it with sleeps.append records each delay instead of waiting, so the test proves the rule from the errors lesson, three waits for four attempts and none after the last, in milliseconds.
Stub programs and a real local server
Code that runs another program is best tested against a stub: a tiny executable with the same name, placed first on PATH, that behaves the way the test needs. The real subprocess.run() call then runs unchanged, arguments, timeout and all. scan_runner.py wraps a scanner called fakescan (the capstone's scanner), using the rules from the subprocess lesson:
"""Run the external scanner (fakescan) on one file and return its findings."""import jsonimport subprocessclass ScanError(Exception):"""No usable answer from the scanner: missing, failed, hung, or not its JSON report."""def scan_file(path: str, timeout: float = 30.0) -> list[dict]:try:result = subprocess.run(["fakescan", "--json", "--", path],capture_output=True, text=True, timeout=timeout,)except FileNotFoundError as err:raise ScanError("fakescan is not installed or not on PATH") from errexcept subprocess.TimeoutExpired as err:raise ScanError(f"fakescan gave no answer within {timeout:g}s") from errif result.returncode != 0:raise ScanError(f"fakescan exited {result.returncode}: {result.stderr.strip()}")try:return json.loads(result.stdout)["findings"]except (json.JSONDecodeError, KeyError) as err:raise ScanError("fakescan printed something that is not its JSON report") from err
"""scan_file() against a stub fakescan on PATH: the real subprocess.run call runs every time."""import osimport pytestfrom scan_runner import ScanError, scan_file@pytest.fixturedef fakescan(tmp_path, monkeypatch):"""Return a function that installs a stub fakescan, first on PATH, running the given sh code."""bin_dir = tmp_path / "bin"bin_dir.mkdir()monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}")def install(sh_code: str) -> None:stub = bin_dir / "fakescan"stub.write_text(f"#!/bin/sh\n{sh_code}\n")stub.chmod(0o755)return installdef test_findings_are_returned(fakescan):fakescan("""echo '{"findings": [{"rule": "test-signature", "severity": "high"}]}'""")assert scan_file("upload.bin") == [{"rule": "test-signature", "severity": "high"}]def test_path_is_passed_after_double_dash(fakescan, tmp_path):fakescan(f"""printf '%s\\n' "$@" > {tmp_path}/argv; echo '{{"findings": []}}'""")scan_file("--delete-all") # a hostile file name that looks like an optionassert (tmp_path / "argv").read_text().split("\n") == ["--json", "--", "--delete-all", ""]def test_scanner_failure(fakescan):fakescan("echo 'signature database missing' >&2; exit 2")with pytest.raises(ScanError, match="exited 2: signature database missing"):scan_file("upload.bin")def test_garbage_output(fakescan):fakescan("echo 'Scanning... done'")with pytest.raises(ScanError, match="not its JSON report"):scan_file("upload.bin")def test_hung_scanner_times_out(fakescan):fakescan("exec sleep 30") # exec: the stub becomes sleep, so the kill on timeout reaches itwith pytest.raises(ScanError, match="no answer within 0.5s"):scan_file("upload.bin", timeout=0.5)def test_missing_scanner(monkeypatch):monkeypatch.setenv("PATH", "/nonexistent")with pytest.raises(ScanError, match="not on PATH"):scan_file("upload.bin")
The fakescan fixture returns a function, so each test writes its own stub script. The second test records the arguments the stub received and proves the file name arrived after --, where a name like --delete-all cannot be read as an option. The hang test uses exec sleep 30: exec replaces the shell with sleep, so when the timeout kills the child it kills the program that is actually waiting, and nothing is left running.
--durations=3 shows the slowest tests: the timeout test took half a second, its own timeout, and the others were too fast to list. For HTTP, a fake function cannot catch mistakes in how the request is made. test_feed.py runs a real http.server in a background thread:
"""feed.get_json() against a real HTTP server on 127.0.0.1, on a port the OS picks."""import threadingimport timefrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerimport pytestfrom feed import PermanentFeedError, TemporaryFeedError, get_jsonROUTES = {"/ok": (200, b'{"verdict": "malicious"}'),"/missing": (404, b'{"error": "no such indicator"}'),"/down": (503, b'{"error": "maintenance"}'),"/html": (200, b"<html>maintenance page</html>"),}class Handler(BaseHTTPRequestHandler):def do_GET(self):if self.path == "/drop": # read the request, then close without answeringself.close_connection = Truereturnif self.path == "/slow":time.sleep(2) # longer than the timeout the test gives the clientstatus, body = ROUTES.get(self.path, (200, b"{}"))try:self.send_response(status)self.send_header("Content-Length", str(len(body)))self.end_headers()self.wfile.write(body)except OSError:pass # the client already gave updef log_message(self, *args): # keep the test output quietpass@pytest.fixture(scope="module") # one server for all tests in this filedef base_url():server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) # port 0: any free portthread = threading.Thread(target=server.serve_forever, daemon=True)thread.start()yield f"http://127.0.0.1:{server.server_address[1]}"server.shutdown() # after the last test of the file: stop and close the socketserver.server_close()def test_json_answer(base_url):assert get_json(base_url + "/ok", timeout=2) == {"verdict": "malicious"}@pytest.mark.parametrize("path, error",[("/missing", PermanentFeedError),("/html", PermanentFeedError),("/down", TemporaryFeedError),("/slow", TemporaryFeedError),("/drop", TemporaryFeedError),],)def test_failure_is_classified(base_url, path, error):with pytest.raises(error):get_json(base_url + path, timeout=0.5)
Binding to port 0 lets the operating system pick a free port, so the tests never collide with another service or a second test run; server_address[1] is the port it chose. The fixture yields the base URL: code before yield is setup, code after it is teardown, and scope="module" starts one server for all tests in the file. /slow answers after two seconds while the client allows 0.5, which checks that a timeout counts as temporary, and /drop closes the connection without an answer, the RemoteDisconnected case from the errors lesson. log_message(self, *args) accepts any number of positional arguments (*args collects them into a tuple) and ignores them. daemon=True makes the server's thread one that does not keep the process alive when the tests are done.
The whole suite, 32 tests, runs in under two seconds, with no network, no real scanner and no files outside temporary directories.
Try this
Write tests/test_duplicate.py with one test that uses workdir and caplog: copy uploads/a/invoice.pdf to uploads/mail/invoice.pdf, run main() with quarantine on both files, and assert exit status 3, that the copy is still in uploads/mail/, and that File exists appears in the log. It passes. Then break the code the test protects: remove | os.O_EXCL from quarantine.py (keep a copy) and run the test again. It fails with assert 1 == 3, and the captured stdout shows both files "moved" to the same destination, the second over the first. Restore quarantine.py and the test passes again.
Takeaway
Test behaviour through the same entry points your users have (main(argv), a real PATH lookup, a real socket), and check effects and exit statuses as well as messages. Patch the name the code under test looks up, and prove failure paths (errors, hangs, bad data) with fakes you control.
report.py starts with from subprocess import run and calls run([...]). A test does monkeypatch.setattr(subprocess, "run", fake). What happens when the test calls report.main()?setattr rebinds one name on one module. report keeps its own reference to the real function.report.run, the name main() looks up, as test_check_feed.py patches check_feed.get_json.time.sleep.report at the top, before any test runs; re-ordering is fragile. Patch the right name instead.test_feed.py bind to port 0 instead of a fixed port such as 18000?127.0.0.1 does that. Port 0 only says which port to use.server_address and builds the URL from it.test_dry_run_changes_nothing compares the list of files before and after --dry-run. A colleague suggests only checking that the output contains "would move". What does that version miss?caplog can check them alongside.