105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
from pathlib import Path
|
|
import os
|
|
import threading
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
def test_web_upload_sample_files():
|
|
client = TestClient(app)
|
|
opened = []
|
|
try:
|
|
files = []
|
|
for path in sorted(Path("PCF文件样本(1)").glob("*.pcf")):
|
|
handle = path.open("rb")
|
|
opened.append(handle)
|
|
files.append(("files", (path.name, handle, "text/plain")))
|
|
|
|
response = client.post("/api/jobs", files=files)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["input_files"] == 20
|
|
assert payload["weld_count"] == 169
|
|
assert len(payload["tables"]["import_rows"]) == 169
|
|
assert {"key": "unit_code", "label": "单位代码"} in payload["tables"]["import_columns"]
|
|
assert "unit_code" in payload["tables"]["import_rows"][0]
|
|
assert "line_no" in payload["tables"]["import_rows"][0]
|
|
assert "单位代码" not in payload["tables"]["import_rows"][0]
|
|
assert payload["tables"]["issue_rows"]
|
|
assert "单位代码" in payload["tables"]["import_headers"]
|
|
|
|
xls = client.get(payload["downloads"]["xls"])
|
|
assert xls.status_code == 200
|
|
assert xls.content[:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"
|
|
finally:
|
|
for handle in opened:
|
|
handle.close()
|
|
|
|
|
|
def test_parse_url_api_returns_json_payload(tmp_path):
|
|
sample = sorted(Path("PCF文件样本(1)").glob("*.pcf"))[0]
|
|
hosted = tmp_path / sample.name
|
|
hosted.write_bytes(sample.read_bytes())
|
|
|
|
class QuietHandler(SimpleHTTPRequestHandler):
|
|
def log_message(self, format, *args):
|
|
return
|
|
|
|
old_cwd = os.getcwd()
|
|
os.chdir(tmp_path)
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
url = f"http://127.0.0.1:{server.server_port}/{hosted.name}"
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
"/api/AnalysisPcf",
|
|
json={"attachment_url": url},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["input_files"] == 1
|
|
assert payload["weld_count"] > 0
|
|
assert payload["tables"]["import_rows"]
|
|
assert "line_no" in payload["tables"]["import_rows"][0]
|
|
assert "管线代号" not in payload["tables"]["import_rows"][0]
|
|
assert "files" not in payload
|
|
assert "downloads" not in payload
|
|
assert payload["report"]["summary"]["weld_count"] == payload["weld_count"]
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
os.chdir(old_cwd)
|
|
|
|
|
|
def test_parse_url_api_accepts_local_file_path():
|
|
sample = sorted(Path("PCF文件样本(1)").glob("*.pcf"))[0].resolve()
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
"/api/AnalysisPcf",
|
|
json={"attachment_url": str(sample)},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["input_files"] == 1
|
|
assert payload["weld_count"] > 0
|
|
assert payload["tables"]["import_rows"]
|
|
|
|
|
|
def test_parse_url_api_accepts_file_uri():
|
|
sample = sorted(Path("PCF文件样本(1)").glob("*.pcf"))[0].resolve()
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
"/api/AnalysisPcf",
|
|
json={"attachment_url": sample.as_uri()},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["input_files"] == 1
|
|
assert payload["weld_count"] > 0
|
|
assert "downloads" not in payload
|