1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
import json
import os
import signal
import socket
import subprocess
from test_general import SalisTest
ANC_SIZE = 12
SERVER_HOST = "localhost"
SERVER_PORT = 8080
SIM_NAME = "def.sim"
SIM_OPTS = {
"anc": "datat",
"anc_path": "test/anc",
"clones": 2,
"cores": 1,
"data_push_pow": 12,
"force": False,
"muta_pow": 32,
"mvec_pow": 8,
"seed": "0x0",
"vm_arch": "v1",
"sync_pow": 10,
"auto_save_pow": 36,
}
class TestDataV1(SalisTest):
def request_from_server(self, request):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
client.connect((SERVER_HOST, SERVER_PORT))
client.sendall(json.dumps(request).encode())
client.shutdown(socket.SHUT_WR)
return json.load(client.makefile(mode="r"))
@SalisTest.run_ui_test(**SIM_OPTS, name=SIM_NAME, ui="push", ui_path="test/ui")
def test_data_flow(self):
with SalisTest.run_pipe(f"./salis.py server -H{self.tempdir.name}") as server_proc:
for line in server_proc.stdout:
if "Listening..." in line:
break
try:
self.assert_equal(self.request_from_server({"request": "name"})["name"], "def.sim")
self.assert_equal(self.request_from_server({"request": "hash"})["hash"], subprocess.check_output("git rev-parse HEAD".split()).decode().strip())
self.assert_equal(self.request_from_server({"request": "opts"}), SIM_OPTS)
data = self.request_from_server({
"request": "data",
"x-axis": "rowid",
"x-current": 0,
"x-high": 2,
"nth": 1,
"entries": 2,
"hm-left": 0,
"hm-pixel-count": 2 ** SIM_OPTS["mvec_pow"],
"hm-pixel-pow": 0,
})
with open(os.path.join(self.tempdir.name, SIM_NAME, "outputs.json"), "r") as f:
outputs = json.load(f)
# Check scalar values
#self.assert_equal(data["rowid"], [1, 2])
#self.assert_equal(data["cycl_0"], [0, 923])
#self.assert_equal(data["mall_0"], [ANC_SIZE * 2, (ANC_SIZE * 2) + 12])
#self.assert_equal(data["pnum_0"], [2, 5])
#self.assert_equal(data["pfst_0"], [0, 0])
#self.assert_equal(data["plst_0"], [1, 4])
#self.assert_equal(data["amb0_0"], [ANC_SIZE, ((ANC_SIZE * 2) + (4 * 3)) // 5])
#self.assert_equal(data["amb1_0"], [0, 0])
#self.assert_equal(data["emb0_0"], [0, 1872])
#self.assert_equal(data["emb1_0"], [0, 2])
#self.assert_equal(data["eliv_0"], [0, 1894])
#self.assert_equal(data["edea_0"], [0, 328])
#self.assert_equal(data["wmb0_0"], [0, 1])
#self.assert_equal(data["wmb1_0"], [0, 0])
#self.assert_equal(data["wdea_0"], [0, 0])
#self.assert_equal(data["step"], [0, 2 ** SIM_OPTS["data_push_pow"]])
print("\ndata ---->\n", json.dumps(data, sort_keys=True))
print("\noutputs ---->\n", json.dumps(outputs, sort_keys=True))
#self.assertDictEqual(data, outputs, "mismatch between server-data and ui outputs")
finally:
server_proc.send_signal(signal.SIGINT)
|
