Something went wrong. Try again.
OR-1 dataflow CPU sketch
Something went wrong. Try again.
6.9 kB · 201 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202"""Tests for dfgraph server WebSocket and file watcher.
Tests verify:- dataflow-renderer.AC4.2: Initial load renders the graph without manual refresh- dataflow-renderer.AC4.1: Saving the dfasm file triggers re-render within 1 second"""
import time
from starlette.testclient import TestClient
from dfgraph.server import create_app
class TestInitialLoad: """AC4.2: Initial load renders the graph without manual refresh."""
def test_websocket_sends_initial_graph_on_connect(self, tmp_path): """Connect to /ws and receive initial graph JSON.""" # Create a minimal valid dfasm file dfasm_file = tmp_path / "test.dfasm" dfasm_file.write_text("""@system pe=2, sm=0&c1|pe0 <| const, 3&c2|pe0 <| const, 7&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""")
app = create_app(dfasm_file) with TestClient(app) as client: with client.websocket_connect("/ws") as ws: data = ws.receive_json()
assert data["type"] == "graph_update" assert "nodes" in data assert "edges" in data assert "metadata" in data # For a valid graph, nodes should be non-empty assert isinstance(data["nodes"], list)
def test_websocket_graph_has_expected_fields(self, tmp_path): """Graph JSON has all expected fields.""" dfasm_file = tmp_path / "test.dfasm" dfasm_file.write_text("""@system pe=2, sm=0&c1|pe0 <| const, 3&c2|pe0 <| const, 7&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""")
app = create_app(dfasm_file) with TestClient(app) as client: with client.websocket_connect("/ws") as ws: data = ws.receive_json()
assert "type" in data assert "stage" in data assert "nodes" in data assert "edges" in data assert "regions" in data assert "errors" in data assert "parse_error" in data assert "metadata" in data
class TestLiveReload: """AC4.1: Saving the dfasm file triggers re-render within 1 second."""
def test_file_change_broadcasts_update(self, tmp_path): """Modify file on disk and receive updated graph.""" dfasm_file = tmp_path / "test.dfasm" dfasm_file.write_text("""@system pe=2, sm=0&c1|pe0 <| const, 3&c2|pe0 <| const, 7&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""")
app = create_app(dfasm_file) with TestClient(app) as client: with client.websocket_connect("/ws") as ws: # Receive initial graph data1 = ws.receive_json() assert data1["type"] == "graph_update" initial_consts = sorted([n["const"] for n in data1["nodes"] if n["opcode"] == "const"])
# Modify the file dfasm_file.write_text("""@system pe=2, sm=0&c1|pe0 <| const, 5&c2|pe0 <| const, 9&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""")
# Wait for update with generous timeout (up to 2 seconds) start = time.time() data2 = None while time.time() - start < 2.0: try: data2 = ws.receive_json() if data2: break except Exception: time.sleep(0.1)
assert data2 is not None, "No update received within 2 seconds" assert data2["type"] == "graph_update" updated_consts = sorted([n["const"] for n in data2["nodes"] if n["opcode"] == "const"]) # Verify the consts actually changed assert updated_consts != initial_consts, f"Graph not actually updated: {initial_consts} vs {updated_consts}"
def test_rapid_file_changes_debounced(self, tmp_path): """Rapid file modifications result in single update (debounce).""" dfasm_file = tmp_path / "test.dfasm" dfasm_file.write_text("""@system pe=2, sm=0&c1|pe0 <| const, 3&c2|pe0 <| const, 7&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""")
app = create_app(dfasm_file) with TestClient(app) as client: with client.websocket_connect("/ws") as ws: # Receive initial graph data1 = ws.receive_json() assert data1["type"] == "graph_update"
# Modify file rapidly 3 times - these should be debounced together for i in range(3): dfasm_file.write_text(f"""@system pe=2, sm=0&c1|pe0 <| const, {3 + i}&c2|pe0 <| const, {7 + i}&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""") time.sleep(0.1)
# Give debounce time to trigger (300ms debounce + buffer) time.sleep(0.5)
# Collect updates within a 2-second window start = time.time() received_updates = [] while time.time() - start < 2.0: try: data = ws.receive_json() if data.get("type") == "graph_update": received_updates.append(data) break except Exception: break
assert len(received_updates) >= 1, ( f"Expected at least 1 debounced update, got {len(received_updates)}" )
class TestHttpServing: """HTTP serving of static files."""
def test_http_get_index_html(self, tmp_path): """GET / returns index.html with dfgraph title.""" dfasm_file = tmp_path / "test.dfasm" dfasm_file.write_text("""@system pe=2, sm=0&c1|pe0 <| const, 3&c2|pe0 <| const, 7&result|pe0 <| add&c1|pe0 |> &result|pe0:L&c2|pe0 |> &result|pe0:R""")
app = create_app(dfasm_file) with TestClient(app) as client: response = client.get("/") assert response.status_code == 200 assert "dfgraph" in response.text
class TestParseError: """Handle invalid dfasm source gracefully."""
def test_parse_error_in_initial_graph(self, tmp_path): """Invalid dfasm produces parse error in graph.""" dfasm_file = tmp_path / "test.dfasm" # Write invalid dfasm (missing system block) dfasm_file.write_text("this is not valid dfasm syntax @#$")
app = create_app(dfasm_file) with TestClient(app) as client: with client.websocket_connect("/ws") as ws: data = ws.receive_json()
assert data["type"] == "graph_update" assert data["parse_error"] is not None or data["stage"] == "parse_error"