Mia.Wu/test_dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: allow network paths

Browse Source
This commit is contained in:
Mia Wu
2026-05-27 15:02:32 -04:00
parent 5996443f38
commit ba86f5830f
9 changed files with 342 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/scanner.py
+147 -32
View File
@@ -1,5 +1,6 @@
import os
import re
import smbclient
from db_py import clear_tests, get_all_tests, mark_completed, set_coe_pair, upsert_test
from parser import (
@@ -11,16 +12,125 @@ from parser import (
)
def full_scan(target_dir, results_dir):
_SMB_SESSIONS = set()
def _normalize_input_path(path_value):
if not path_value:
return path_value
path = str(path_value).strip()
# Accept //server/share style and normalize to UNC for smbclient.
if path.startswith("//"):
return "\\\\" + path.lstrip("/").replace("/", "\\")
# Accept /<ipv4>/<share>/... and normalize to UNC for Linux-hosted inputs.
if re.match(r"^/\d{1,3}(?:\.\d{1,3}){3}/[^/]+", path):
return "\\\\" + path.lstrip("/").replace("/", "\\")
return path
def _is_unc_path(path):
return isinstance(path, str) and path.startswith("\\\\")
def _extract_unc_server(path):
if not _is_unc_path(path):
return None
rest = path[2:]
return rest.split("\\", 1)[0] if rest else None
def _register_smb_session_if_needed(path):
if not _is_unc_path(path):
return
server = _extract_unc_server(path)
if not server or server in _SMB_SESSIONS:
return
username = os.getenv("SMB_USERNAME", "").strip()
password = os.getenv("SMB_PASSWORD", "")
domain = os.getenv("SMB_DOMAIN", "").strip()
if username and domain and "\\" not in username and "@" not in username:
username = f"{domain}\\{username}"
if username:
smbclient.register_session(server, username=username, password=password)
else:
smbclient.register_session(server)
_SMB_SESSIONS.add(server)
def _iter_dir_entries(path):
path = _normalize_input_path(path)
if _is_unc_path(path):
_register_smb_session_if_needed(path)
return list(smbclient.scandir(path))
return list(os.scandir(path))
def _join_path(path, name):
if _is_unc_path(path):
base = path.rstrip("\\")
return f"{base}\\{name}"
return os.path.join(path, name)
def _read_text_lines(path):
path = _normalize_input_path(path)
if _is_unc_path(path):
_register_smb_session_if_needed(path)
with smbclient.open_file(path, mode="r", encoding="utf-8", errors="ignore") as fh:
for line in fh:
yield line
return
with open(path, "r", encoding="utf-8", errors="ignore") as fh:
for line in fh:
yield line
def scan_results_only(results_dir, results_dir_ref=None):
"""Re-scan only the results directories without clearing or re-scanning targets."""
results_dir = _normalize_input_path(results_dir)
results_dir_ref = _normalize_input_path(results_dir_ref)
if not results_dir:
return
print(f"[scanner] results-only scan dir : {results_dir}")
if results_dir_ref:
print(f"[scanner] results-only scan dir ref : {results_dir_ref}")
scan_results(results_dir)
if results_dir_ref:
scan_results(results_dir_ref)
update_p2p_coe_pairs()
def full_scan(target_dir, results_dir, results_dir_ref=None):
target_dir = _normalize_input_path(target_dir)
results_dir = _normalize_input_path(results_dir)
results_dir_ref = _normalize_input_path(results_dir_ref)
clear_tests()
if not target_dir or not results_dir:
return
print(f"[scanner] target dir : {target_dir}")
print(f"[scanner] results dir: {results_dir}")
print(f"[scanner] target dir : {target_dir}")
print(f"[scanner] results dir : {results_dir}")
if results_dir_ref:
print(f"[scanner] results dir ref : {results_dir_ref}")
scan_targets(target_dir)
scan_results(results_dir)
if results_dir_ref:
scan_results(results_dir_ref)
update_p2p_coe_pairs()
@@ -66,11 +176,13 @@ def update_p2p_coe_pairs():
def scan_targets(target_dir):
target_dir = _normalize_input_path(target_dir)
try:
parent_entries = [
name
for name in os.listdir(target_dir)
if os.path.isdir(os.path.join(target_dir, name))
entry.name
for entry in _iter_dir_entries(target_dir)
if entry.is_dir()
]
except OSError as exc:
print(f"[scanner] Cannot read target dir: {exc}")
@@ -79,14 +191,14 @@ def scan_targets(target_dir):
print(f"[scanner] subdirectories found: {len(parent_entries)}")
for parent_name in parent_entries:
parent_path = os.path.join(target_dir, parent_name)
parent_path = _join_path(target_dir, parent_name)
try:
files = [
name
for name in os.listdir(parent_path)
if os.path.isfile(os.path.join(parent_path, name))
and not name.startswith("GLOBAL")
and name.endswith(".ini")
entry.name
for entry in _iter_dir_entries(parent_path)
if entry.is_file()
and not entry.name.startswith("GLOBAL")
and entry.name.endswith(".ini")
]
except OSError as exc:
print(f"[scanner] Cannot read parent dir {parent_name}: {exc}")
@@ -125,11 +237,13 @@ def scan_targets(target_dir):
def scan_results(results_dir):
results_dir = _normalize_input_path(results_dir)
try:
entries = [
name
for name in os.listdir(results_dir)
if os.path.isdir(os.path.join(results_dir, name))
entry.name
for entry in _iter_dir_entries(results_dir)
if entry.is_dir()
]
except OSError as exc:
print(f"[scanner] Cannot read results dir: {exc}")
@@ -141,21 +255,23 @@ def scan_results(results_dir):
def process_result_dir(results_dir, result_dir_name):
results_dir = _normalize_input_path(results_dir)
parsed = parse_result_filename(result_dir_name)
test_id = parsed["test_id"]
device = parsed["device"]
if not test_id or not device:
return
result_dir_path = os.path.join(results_dir, result_dir_name)
result_dir_path = _join_path(results_dir, result_dir_name)
try:
log_files = [
name
for name in os.listdir(result_dir_path)
if os.path.isfile(os.path.join(result_dir_path, name))
and name.endswith(".txt")
and test_id in name
entry.name
for entry in _iter_dir_entries(result_dir_path)
if entry.is_file()
and entry.name.endswith(".txt")
and test_id in entry.name
]
except OSError as exc:
print(f"[scanner] Cannot read result dir {result_dir_name}: {exc}")
@@ -167,7 +283,7 @@ def process_result_dir(results_dir, result_dir_name):
return
latest_log = sorted(log_files)[-1]
log_path = os.path.join(result_dir_path, latest_log)
log_path = _join_path(result_dir_path, latest_log)
completed_at = parse_timestamp(latest_log)
data = extract_log_data(log_path)
@@ -199,17 +315,16 @@ def extract_log_data(log_file_path):
result = {"duration_seconds": None, "tputResults": []}
try:
with open(log_file_path, "r", encoding="utf-8", errors="ignore") as fh:
for line in fh:
time_match = elapsed_regex.search(line)
if time_match:
result["duration_seconds"] = parse_elapsed_time(time_match.group(1))
continue
for line in _read_text_lines(log_file_path):
time_match = elapsed_regex.search(line)
if time_match:
result["duration_seconds"] = parse_elapsed_time(time_match.group(1))
continue
if tput_regex.search(line):
parsed = parse_tput_rssi(line)
if parsed:
result["tputResults"].append(parsed)
if tput_regex.search(line):
parsed = parse_tput_rssi(line)
if parsed:
result["tputResults"].append(parsed)
except OSError as exc:
print(f"[scanner] Cannot read log file {log_file_path}: {exc}")