184 lines
8.0 KiB
Python
184 lines
8.0 KiB
Python
"""SD provisioning: ENV.TXT (firmware-exact parsing), disk discovery parsers,
|
|
the elevated Windows format script, and write/check on a stand-in card."""
|
|
|
|
import plistlib
|
|
|
|
import pytest
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from kota_tech.sd.disks import MAX_CARD_BYTES, Disk, Volume, parse_macos, parse_windows
|
|
from kota_tech.sd.envfile import EnvConfig, parse_env
|
|
from kota_tech.sd.provision import check_card, win_format_command, write_env
|
|
|
|
URL = "https://influx.kota-app.com/api/v2/write?bucket=sensorbox&org=kota"
|
|
TOKEN = "T" * 86 + "==" # shape of an InfluxDB token, not a real one
|
|
|
|
|
|
def cfg(**kw):
|
|
base = dict(influx_url=URL, influx_token=TOKEN)
|
|
base.update(kw)
|
|
return EnvConfig(**base)
|
|
|
|
|
|
# ------------------------------------------------------------------ ENV.TXT
|
|
|
|
def test_render_matches_the_factory_file_format():
|
|
raw = cfg().render()
|
|
assert raw == f"INFLUX_TOKEN={TOKEN}\nINFLUX_URL={URL}\n".encode()
|
|
assert not raw.startswith(b"\xef\xbb\xbf") and b"\r" not in raw
|
|
assert parse_env(raw).ok
|
|
|
|
|
|
def test_wifi_only_written_when_both_set():
|
|
raw = cfg(ssid="Site WiFi", password="correct horse").render().decode()
|
|
assert "SSID=Site WiFi\n" in raw and "PASSWORD=correct horse\n" in raw
|
|
assert any("both SSID and PASSWORD" in i.text for i in cfg(ssid="x").problems())
|
|
|
|
|
|
@pytest.mark.parametrize("kw,fragment", [
|
|
(dict(influx_url="influx.kota-app.com"), "https://"),
|
|
(dict(influx_url="https://influx.kota-app.com/"), "write URL"),
|
|
(dict(influx_token=""), "INFLUX_TOKEN is empty"),
|
|
(dict(influx_token="abc def" * 5), "spaces"),
|
|
(dict(influx_token=TOKEN + " "), "space"),
|
|
(dict(ssid="s" * 33, password="12345678"), "32 bytes"),
|
|
(dict(ssid="net", password="short"), "8-63"),
|
|
])
|
|
def test_invalid_settings_are_errors(kw, fragment):
|
|
errors = [i.text for i in cfg(**kw).problems() if i.level == "error"]
|
|
assert any(fragment in e for e in errors), errors
|
|
|
|
|
|
def test_parser_mirrors_firmware():
|
|
raw = b"\xef\xbb\xbfINFLUX_TOKEN=abc\r\nINFLUX_URL = x\nINFLUX_URL=u1\nINFLUX_URL=u2\nHTTP_SERVER=true\n"
|
|
chk = parse_env(raw)
|
|
texts = " ".join(i.text for i in chk.issues)
|
|
assert "BOM" in texts # first key unreadable
|
|
assert "spaces around the key 'INFLUX_URL'" in texts
|
|
assert chk.values["INFLUX_URL"] == "u1" # first match wins
|
|
assert "appears twice" in texts and "setup mode" in texts
|
|
assert "INFLUX_TOKEN is missing" in texts # hidden behind the BOM
|
|
assert not chk.ok
|
|
|
|
|
|
def test_masking():
|
|
chk = parse_env(cfg(ssid="net", password="secretpassword").render())
|
|
m = chk.masked()
|
|
assert TOKEN not in m["INFLUX_TOKEN"] and m["INFLUX_TOKEN"].startswith("TTTT")
|
|
assert "secret" not in m["PASSWORD"]
|
|
|
|
|
|
# ------------------------------------------------------------- disk lists
|
|
|
|
WIN_JSON = """[{"Number":0,"FriendlyName":"ST500DM002-1BC142","Size":500107862016,"BusType":"SATA",
|
|
"Volumes":[{"DriveLetter":"D","FileSystem":"NTFS","Label":"WINRE"}]},
|
|
{"Number":4,"FriendlyName":"Mass Storage Device","Size":0,"BusType":"USB","Volumes":[]},
|
|
{"Number":5,"FriendlyName":"SDXC Card","Size":31914983424,"BusType":"SD",
|
|
"Volumes":{"DriveLetter":"E","FileSystem":"FAT32","Label":"KOTA"}}]"""
|
|
|
|
|
|
def test_parse_windows_keeps_only_card_buses():
|
|
disks = parse_windows(WIN_JSON)
|
|
assert [d.id for d in disks] == ["4", "5"] # SATA data disk never listed
|
|
reader, card = disks
|
|
assert not reader.has_media and reader.size_text == "no card"
|
|
assert card.volumes[0].mount == "E:\\" and card.volumes[0].filesystem == "FAT32"
|
|
|
|
|
|
def test_parse_windows_single_object_quirk():
|
|
one = '{"Number":5,"FriendlyName":"SD","Size":8000000000,"BusType":"USB","Volumes":null}'
|
|
assert [d.id for d in parse_windows(one)] == ["5"]
|
|
|
|
|
|
def test_parse_macos_excludes_system_disk():
|
|
listing = plistlib.dumps({"AllDisksAndPartitions": [
|
|
{"DeviceIdentifier": "disk0", "Size": 500_000_000_000, "Partitions": []},
|
|
{"DeviceIdentifier": "disk3", "Size": 994_000_000_000, "Partitions": [], "APFSVolumes": [{"MountPoint": "/"}]},
|
|
{"DeviceIdentifier": "disk4", "Size": 15_931_539_456,
|
|
"Partitions": [{"DeviceIdentifier": "disk4s1", "MountPoint": "/Volumes/KOTA", "Content": "DOS_FAT_32",
|
|
"VolumeName": "KOTA"}]},
|
|
]})
|
|
infos = {"disk0": plistlib.dumps({"Internal": True, "BusProtocol": "PCI-Express"}),
|
|
"disk3": plistlib.dumps({"Internal": True}),
|
|
"disk4": plistlib.dumps({"Internal": True, "RemovableMedia": True, "BusProtocol": "Secure Digital",
|
|
"MediaName": "SD Card Reader"})}
|
|
disks = parse_macos(listing, infos)
|
|
assert [d.id for d in disks] == ["/dev/disk4"] and disks[0].bus == "SD"
|
|
assert disks[0].volumes[0].mount == "/Volumes/KOTA"
|
|
|
|
|
|
# -------------------------------------------------------- format script
|
|
|
|
def test_elevated_helper_command(tmp_path):
|
|
disk = Disk("5", "SDXC Card", 63_864_569_856, "SD")
|
|
prog, params, cwd = win_format_command(disk, tmp_path / "r.json")
|
|
assert params.startswith("-m kota_tech.sd.win_format --disk 5 --size 63864569856 --label KOTA")
|
|
assert (Path(cwd) / "kota_tech").is_dir() # helper is importable from there
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "win32", reason="Windows formatter")
|
|
@pytest.mark.parametrize("disks,msg", [
|
|
([], "not a USB/SD/MMC"), # disk vanished / internal
|
|
([Disk("5", "SD", 1_000, "SD")], "changed since it was selected"), # swapped card
|
|
([Disk("5", "SD", 31_457_280_000, "SD", [Volume("C:\\", "NTFS", "")])], "Windows drive"),
|
|
])
|
|
def test_formatter_guard_refuses_before_writing(monkeypatch, disks, msg):
|
|
from kota_tech.sd import win_format
|
|
monkeypatch.setattr(win_format.win_disks, "list_disks", lambda buses: disks)
|
|
monkeypatch.setattr(win_format, "RawDisk", lambda *a: pytest.fail("must not open the disk for writing"))
|
|
with pytest.raises(win_format.FormatError, match=msg):
|
|
win_format.format_card(5, 31_457_280_000, "KOTA", {})
|
|
|
|
|
|
def test_large_disks_are_not_cards():
|
|
assert MAX_CARD_BYTES == 256 * 1024**3
|
|
|
|
|
|
# ----------------------------------------------- write / check on a card
|
|
|
|
def test_write_env_replaces_variants_and_keeps_buffered_data(tmp_path):
|
|
card = tmp_path / "card"
|
|
card.mkdir()
|
|
(card / "env.txt").write_bytes(b"INFLUX_URL=old\n")
|
|
(card / "BUFFER.TXT").write_bytes(b"line1\nline2\n\n")
|
|
out = write_env(str(card), cfg())
|
|
assert out.ok, out.message
|
|
names = sorted(p.name for p in card.iterdir())
|
|
assert names == ["BUFFER.TXT", "ENV.TXT"]
|
|
rep = check_card(str(card), "FAT32")
|
|
assert rep.env.ok and rep.problems == [] and rep.buffered_lines == 2
|
|
|
|
|
|
def test_write_env_refuses_invalid_settings(tmp_path):
|
|
assert not write_env(str(tmp_path), cfg(influx_token="")).ok
|
|
assert not (tmp_path / "ENV.TXT").exists()
|
|
|
|
|
|
def test_check_card_reports_missing_env_and_wrong_fs(tmp_path):
|
|
rep = check_card(str(tmp_path), "exFAT")
|
|
texts = " ".join(rep.problems)
|
|
assert "needs FAT32" in texts and "No ENV.TXT" in texts
|
|
|
|
|
|
def test_elevated_result_folder_is_not_owner_only(tmp_path, monkeypatch):
|
|
# Regression: tempfile.mkdtemp() made the elevated script's result unreadable.
|
|
import kota_tech.sd.provision as prov
|
|
monkeypatch.setattr(prov, "WORK_ROOT", tmp_path / "tmp")
|
|
work = prov._workdir()
|
|
assert work.parent == tmp_path / "tmp" and work.is_dir()
|
|
(work / "result.json").write_text('{"ok": true}', encoding="utf-8")
|
|
assert (work / "result.json").read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.parametrize("fs", ["FAT32", "DOS_FAT_32", "Windows_FAT_32", "MS-DOS FAT32"])
|
|
def test_fat32_names_from_windows_and_macos_are_accepted(tmp_path, fs):
|
|
(tmp_path / "ENV.TXT").write_bytes(b"INFLUX_URL=https://x\nINFLUX_TOKEN=abc\n")
|
|
assert not any("File system" in p for p in check_card(str(tmp_path), fs).problems)
|
|
|
|
|
|
def test_exfat_is_flagged(tmp_path):
|
|
(tmp_path / "ENV.TXT").write_bytes(b"INFLUX_URL=https://x\nINFLUX_TOKEN=abc\n")
|
|
assert any("File system" in p for p in check_card(str(tmp_path), "ExFAT").problems)
|