78 lines
3.0 KiB
Python
78 lines
3.0 KiB
Python
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
SCRIPT = os.path.join(HERE, "shadow_lint.py")
|
|
|
|
|
|
def run(src):
|
|
# Each case gets its own directory. The fixture used to be one fixed path,
|
|
# /tmp/scratchpad/test_shadow.py, shared by every process on the host: two
|
|
# concurrent runs of this gate (another lane, a second CHECK) overwrote
|
|
# each other's source between the write and shadow_lint's read, so a case
|
|
# graded a neighbour's fixture -- measured 10/10 concurrent rounds red,
|
|
# with e.g. the import case reporting the loop-target case's finding.
|
|
with tempfile.TemporaryDirectory(prefix="shadow_case_") as d:
|
|
p = os.path.join(d, "case.py")
|
|
with open(p, "w") as fh:
|
|
fh.write(src)
|
|
r = subprocess.run([sys.executable, SCRIPT, p], capture_output=True, text=True)
|
|
return r.returncode, r.stdout + r.stderr
|
|
|
|
def test_param_shadow():
|
|
rc, out = run("x = 1\ndef f(x):\n return x\n")
|
|
assert rc == 1
|
|
assert "function-param" in out and "'x'" in out
|
|
|
|
def test_no_shadow():
|
|
rc, out = run("x = 1\ndef f(y):\n return x + y\n")
|
|
assert rc == 0, out
|
|
|
|
def test_import_shadow():
|
|
rc, out = run("import os\ndef f():\n import os\n return os\n")
|
|
assert rc == 1 and "import" in out
|
|
|
|
def test_loop_target_shadow():
|
|
rc, out = run("def outer():\n i = 0\n def inner():\n for i in range(3):\n pass\n")
|
|
assert rc == 1 and "loop-target" in out
|
|
|
|
def test_comprehension_shadow():
|
|
rc, out = run("x = 1\ny = [x for x in range(3)]\n")
|
|
assert rc == 1 and "comprehension" in out
|
|
|
|
def test_with_as_shadow():
|
|
rc, out = run("def outer():\n f = 1\n def inner():\n with open('/dev/null') as f:\n pass\n")
|
|
assert rc == 1 and "with-as" in out
|
|
|
|
def test_fixture_isolation():
|
|
"""Regression: the fixture path must be per call, not one host-wide file.
|
|
|
|
With the old fixed /tmp/scratchpad path, two gate runs in the same second
|
|
graded each other's source -- measured 10/10 concurrent rounds red, e.g.
|
|
the import case reporting `shadowed name 'i' (loop-target)`.
|
|
"""
|
|
import inspect, threading
|
|
assert "TemporaryDirectory" in inspect.getsource(run), "run() needs a per-call temp dir"
|
|
got = {}
|
|
cases = {
|
|
"import": "import os\ndef f():\n import os\n return os\n",
|
|
"with-as": "def outer():\n f = 1\n def inner():\n with open('/dev/null') as f:\n pass\n",
|
|
}
|
|
def one(key, src):
|
|
got[key] = run(src)
|
|
threads = [threading.Thread(target=one, args=(k, s)) for k, s in cases.items()]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert got["import"][0] == 1 and "import" in got["import"][1]
|
|
assert got["with-as"][0] == 1 and "with-as" in got["with-as"][1]
|
|
|
|
if __name__ == "__main__":
|
|
for name in [n for n in globals() if n.startswith("test_")]:
|
|
globals()[name]()
|
|
print("PASS", name)
|
|
print("ALL PASS")
|