86 lines
2.2 KiB
Python
86 lines
2.2 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
|
|
PORT = os.environ.get("TELEGRAM_SCRAPER_SMOKE_PORT", "18080")
|
|
BASE_URL = f"http://127.0.0.1:{PORT}"
|
|
ENDPOINTS = [
|
|
"/",
|
|
"/viewer",
|
|
"/swagger",
|
|
"/openapi.json",
|
|
"/health",
|
|
"/health/continuous",
|
|
"/api/dashboard",
|
|
"/api/continuous",
|
|
"/api/channels",
|
|
]
|
|
|
|
|
|
def fetch(path: str) -> tuple[int, bytes]:
|
|
with urllib.request.urlopen(BASE_URL + path, timeout=3) as response:
|
|
return response.status, response.read()
|
|
|
|
|
|
def wait_for_server(process: subprocess.Popen) -> None:
|
|
deadline = time.time() + 15
|
|
last_error = None
|
|
while time.time() < deadline:
|
|
if process.poll() is not None:
|
|
raise RuntimeError(f"server exited with code {process.returncode}")
|
|
try:
|
|
fetch("/health")
|
|
return
|
|
except Exception as exc:
|
|
last_error = exc
|
|
time.sleep(0.3)
|
|
raise RuntimeError(f"server did not become ready: {last_error}")
|
|
|
|
|
|
def main() -> int:
|
|
env = {
|
|
**os.environ,
|
|
"TELEGRAM_SCRAPER_PORT": PORT,
|
|
"TELEGRAM_SCRAPER_START_CONTINUOUS": "0",
|
|
}
|
|
process = subprocess.Popen(
|
|
[sys.executable, "main.py"],
|
|
env=env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
)
|
|
try:
|
|
wait_for_server(process)
|
|
for endpoint in ENDPOINTS:
|
|
status, body = fetch(endpoint)
|
|
if status != 200:
|
|
raise RuntimeError(f"{endpoint} returned HTTP {status}")
|
|
if (
|
|
endpoint.endswith(".json")
|
|
or endpoint.startswith("/api")
|
|
or endpoint.startswith("/health")
|
|
):
|
|
json.loads(body.decode("utf-8"))
|
|
print(f"ok {endpoint}")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"smoke test failed: {exc}", file=sys.stderr)
|
|
if process.stdout:
|
|
print(process.stdout.read(), file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|