829f254194
Exposes pm3py as a pyws workspace, discovered via the pyws.plugins entry point: - ReaderResource: connects/disconnects a Proxmark3 as `reader`; a missing device is non-fatal (reports disconnected) - namespace injection of the transponder/sim classes, SimSession and Proxmark3 - effect declarations so autoreplay/replay withhold writes/field/sim/destructive while reads and pure model edits replay Hardware-free tests via AsyncMock. The live `sim` resource is P2. See docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""Tests for the pm3py pyws plugin (P1: reader resource + namespace + effects). Hardware-free."""
|
|
import asyncio
|
|
from unittest.mock import AsyncMock
|
|
|
|
from pyws import ResourceStatus, WorkspacePlugin
|
|
|
|
from pm3py.pyws_plugin import Pm3pyPlugin, ReaderResource
|
|
|
|
|
|
def _run(coro):
|
|
# match pm3py's suite convention — a shared loop that isn't closed between tests
|
|
# (asyncio.run() closes the loop and poisons every test that runs after this file)
|
|
return asyncio.get_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def test_plugin_is_a_workspace_plugin():
|
|
assert issubclass(Pm3pyPlugin, WorkspacePlugin)
|
|
assert Pm3pyPlugin.name == "pm3py"
|
|
|
|
|
|
def test_entry_point_registered():
|
|
# pyws discovers plugins from this group; requires an (editable) install of pm3py
|
|
from importlib.metadata import entry_points
|
|
|
|
eps = entry_points()
|
|
group = eps.select(group="pyws.plugins") if hasattr(eps, "select") else eps.get("pyws.plugins", [])
|
|
assert "pm3py" in {ep.name for ep in group}
|
|
|
|
|
|
def test_namespace_injects_tag_classes():
|
|
ns = Pm3pyPlugin().namespace(session=None)
|
|
assert "SimSession" in ns
|
|
assert "NTAG213" in ns and callable(ns["NTAG213"])
|
|
assert "Proxmark3" in ns
|
|
# a constructed tag really is a pm3py transponder
|
|
tag = ns["NTAG213"]()
|
|
assert hasattr(tag, "uid")
|
|
|
|
|
|
def test_effects_cover_reader_and_sim():
|
|
eff = Pm3pyPlugin.effects
|
|
assert any("wrbl" in p for p in eff["write"])
|
|
assert any("scan" in p for p in eff["read"])
|
|
assert "sim.start" in eff["sim"]
|
|
assert any(p == "*.sync" for p in eff["write"]) # tag.sync() is a write
|
|
|
|
|
|
def test_effects_classify_nested_pm3py_calls():
|
|
# end-to-end: the declared patterns must actually classify pm3py's nested API
|
|
from pyws.effects import EffectResolver, classify_statement
|
|
|
|
r = EffectResolver()
|
|
r.add_plugin(Pm3pyPlugin())
|
|
assert classify_statement("reader.hf.mfu.wrbl(4, b'aa')", {}, r) == "write"
|
|
assert classify_statement("reader.hf.iso14a.scan()", {}, r) == "read"
|
|
assert classify_statement("authorized_tag.sync()", {}, r) == "write"
|
|
# a pure model edit is unmatched -> unknown -> replays under safe-by-default
|
|
assert classify_statement("authorized_tag.set_page(4, b'aa')", {}, r) == "unknown"
|
|
|
|
|
|
def test_reader_resource_connects_and_disconnects():
|
|
mock = AsyncMock()
|
|
r = ReaderResource("reader", client=mock)
|
|
_run(r.start())
|
|
assert r.status is ResourceStatus.CONNECTED
|
|
assert r.obj is mock
|
|
mock.connect.assert_awaited_once()
|
|
_run(r.stop())
|
|
mock.disconnect.assert_awaited_once()
|
|
assert r.status is ResourceStatus.DISCONNECTED
|
|
|
|
|
|
def test_reader_resource_no_device_is_nonfatal(monkeypatch):
|
|
import pm3py.core.client as client
|
|
|
|
monkeypatch.setattr(client, "find_proxmark_port", lambda: None)
|
|
r = ReaderResource("reader") # no client, no port -> Proxmark3(None) raises -> disconnected
|
|
_run(r.start())
|
|
assert r.status is ResourceStatus.DISCONNECTED
|
|
assert r.obj is None
|
|
|
|
|
|
def test_create_resource_builds_reader():
|
|
assert isinstance(Pm3pyPlugin().create_resource("reader", {}, session=None), ReaderResource)
|
|
assert Pm3pyPlugin().create_resource("sim", {}, session=None) is None # P2
|