Files
Kavosh/tests/test_background_processing.py
2026-08-07 21:17:17 +03:30

53 lines
1.9 KiB
Python

"""Regression coverage for the no-cron simplification (Chapter 12): a
real upload through the HTTP endpoint should reach status "done" on its
own, via the automatic background thread, without any `flask
process-logs` invocation — cron/CLI is optional, not required.
"""
from __future__ import annotations
import io
import time
from app.models.bot_hit import BotHit
from app.models.log_entry import LogEntry
from app.models.log_file import LogFile
SAMPLE_LOG = (
b'203.0.113.10 - - [15/Jul/2026:10:00:00 -0700] "GET /index.html HTTP/1.1" 200 512 '
b'"https://example.com/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0"\n'
b'203.0.113.11 - - [15/Jul/2026:10:00:05 -0700] "GET /about HTTP/1.1" 200 300 '
b'"-" "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)"\n'
)
def test_upload_processes_automatically_without_cron(logged_in_client):
resp = logged_in_client.post(
"/uploads",
data={
"logfile": (io.BytesIO(SAMPLE_LOG), "sample.log"),
"server_type": "apache",
"format_string": "",
},
content_type="multipart/form-data",
)
assert resp.status_code == 200
log_file = LogFile.query.order_by(LogFile.id.desc()).first()
assert log_file is not None
# No `flask process-logs` call anywhere in this test — only the
# background thread triggered by the upload itself should advance it.
deadline = time.time() + 5
while time.time() < deadline:
from app.extensions import db
db.session.refresh(log_file)
if log_file.status in ("done", "error"):
break
time.sleep(0.1)
assert log_file.status == "done", f"expected done, got {log_file.status}: {log_file.error_message}"
assert log_file.total_lines == 2
assert log_file.processed_lines == 2
assert LogEntry.query.filter_by(log_file_id=log_file.id).count() == 2
assert BotHit.query.filter_by(log_file_id=log_file.id).count() == 1