697 lines
23 KiB
Python
697 lines
23 KiB
Python
"""Step definitions for semantic validation scenarios.
|
|
|
|
Covers SemanticValidationService, built-in rules, rule registry,
|
|
cache, severity mapping, config keys, and pipeline integration.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from behave import given, then, when
|
|
|
|
if TYPE_CHECKING:
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.services.semantic_validation_rules import (
|
|
APIMisuseRule,
|
|
BrokenReferenceRule,
|
|
DependencyCycleRule,
|
|
DuplicateImportRule,
|
|
MissingImportRule,
|
|
MissingSymbolRule,
|
|
SemanticCheckResult,
|
|
SemanticValidationSeverity,
|
|
SyntaxCheckRule,
|
|
)
|
|
from cleveragents.application.services.semantic_validation_service import (
|
|
CONFIG_KEY_ENABLED,
|
|
CONFIG_KEY_PYTHON_ENABLED,
|
|
CONFIG_KEY_SEVERITY_MAPPING,
|
|
DEFAULT_CONFIG,
|
|
SemanticValidationCache,
|
|
SemanticValidationService,
|
|
create_default_registry,
|
|
map_severity_to_mode,
|
|
resolve_severity,
|
|
)
|
|
|
|
# -- Fixtures ----------------------------------------------------------------
|
|
VALID_PY = "import os\n\nx = 10\n\ndef hello():\n return x\n"
|
|
SYNTAX_ERR_PY = "def broken(\n return 42\n"
|
|
SUSPICIOUS_IMPORT_PY = "import _nonexistent_private_module\n"
|
|
STD_IMPORT_PY = "import os\nimport sys\nimport json\n"
|
|
ALL_DEFINED_PY = "import os\n\nx = 10\n\ndef foo():\n return x\n"
|
|
UNDEF_REF_PY = "def foo():\n return bar_undefined_xyz\n"
|
|
UNIQUE_REL_PY = "from . import alpha\nfrom . import beta\n"
|
|
DUP_REL_PY = "from . import alpha\nfrom . import alpha\n"
|
|
CLEAN_PY = "import json\ndata = json.loads('{}')\n"
|
|
EVAL_PY = "result = eval('1 + 2')\n"
|
|
SCOPED_PY = "import os\n\ndef greet(name):\n msg = f'Hi {name}'\n return msg\n"
|
|
UNDEF_SYM_PY = "def compute(x):\n return x + undefined_thing_xyz\n"
|
|
FUNC_LOCAL_PY = "def process(items):\n result = []\n for item in items:\n result.append(item)\n return result\n"
|
|
CLASS_METHOD_PY = (
|
|
"class MyClass:\n def method(self):\n return undefined_xyz\n"
|
|
)
|
|
NESTED_FUNC_PY = (
|
|
"def outer():\n def inner():\n return 1\n return inner()\n"
|
|
)
|
|
ANNASSIGN_PY = "x: int = 10\n\ndef foo():\n return x\n"
|
|
WITH_STMT_PY = "import io\ndef read_file():\n with io.open('test.txt') as f:\n return f.read()\n"
|
|
EXEC_PY = "exec('print(1)')\n"
|
|
OS_POPEN_PY = "import os\nos.popen('ls')\n"
|
|
SUBPROCESS_RUN_PY = "import subprocess\nsubprocess.run(['echo', 'hello'])\n"
|
|
COMPREHENSION_PY = "def summarise(items):\n return [x * 2 for x in items]\n"
|
|
NON_PY_FILENAME = "config.yaml"
|
|
EMPTY_PY = ""
|
|
|
|
# -- Fixtures for coverage improvement ----------------------------------------
|
|
FROM_PRIVATE_IMPORT_PY = "from _nonexistent_private_mod import something\n"
|
|
VARARGS_KWARGS_PY = (
|
|
"def func(*args, **kwargs):\n return args, kwargs\n\nfunc(1, key=2)\n"
|
|
)
|
|
FROM_IMPORT_BINDINGS_PY = "from os.path import join\n\nresult = join('a', 'b')\n"
|
|
EXCEPT_HANDLER_PY = "try:\n x = 1\nexcept Exception as exc:\n y = exc\n"
|
|
TUPLE_STARRED_PY = "a, b = 1, 2\nc, *d = [3, 4, 5]\nresult = a + b + c\n"
|
|
WITH_NO_AS_PY = "import warnings\nwith warnings.catch_warnings():\n x = 1\n"
|
|
SAFE_FLAGGED_MODULE_PY = "import os\nresult = os.getcwd()\n"
|
|
FUNC_ALL_PARAMS_PY = (
|
|
"def func(a, /, b, *args, c=1, **kwargs):\n"
|
|
" return a + b + c + len(args) + len(kwargs)\n"
|
|
"\nfunc(1, 2, 3, c=4, d=5)\n"
|
|
)
|
|
FUNC_FOR_LOOP_PY = (
|
|
"def total(items):\n"
|
|
" s = 0\n"
|
|
" for item in items:\n"
|
|
" s += item\n"
|
|
" return s\n"
|
|
)
|
|
FUNC_WITH_STMT_PY = (
|
|
"def reader(path):\n with open(path) as fh:\n return fh.read()\n"
|
|
)
|
|
FUNC_EXCEPT_PY = (
|
|
"def safe_div(a, b):\n"
|
|
" try:\n"
|
|
" return a / b\n"
|
|
" except ZeroDivisionError as err:\n"
|
|
" return err\n"
|
|
)
|
|
FUNC_LOCAL_IMPORT_PY = "def get_cwd():\n import os\n return os.getcwd()\n"
|
|
FUNC_LOCAL_FROM_IMPORT_PY = (
|
|
"def get_path():\n from os.path import join\n return join('a', 'b')\n"
|
|
)
|
|
|
|
|
|
# -- Background --------------------------------------------------------------
|
|
@given("a semantic validation test environment")
|
|
def step_env(context: Context) -> None:
|
|
context.sv_result = None
|
|
context.sv_results = None
|
|
context.sv_source = ""
|
|
context.sv_filename = "test.py"
|
|
context.sv_service = None
|
|
context.sv_cache = None
|
|
context.sv_registry = None
|
|
context.sv_pipeline_results = None
|
|
context.sv_normalised = None
|
|
context.sv_custom_mapping = None
|
|
|
|
|
|
# -- Severity enum -----------------------------------------------------------
|
|
@then('the semantic severity enum has values "info", "warn", "error"')
|
|
def step_sev_vals(context: Context) -> None:
|
|
assert SemanticValidationSeverity.INFO == "info"
|
|
assert SemanticValidationSeverity.WARN == "warn"
|
|
assert SemanticValidationSeverity.ERROR == "error"
|
|
|
|
|
|
# -- SemanticCheckResult model -----------------------------------------------
|
|
@given('a semantic check result with passed true and message "{msg}"')
|
|
def step_cr_pass(context: Context, msg: str) -> None:
|
|
context.sv_result = SemanticCheckResult(passed=True, message=msg)
|
|
|
|
|
|
@given('a semantic check result with passed false and message "{msg}" and data')
|
|
def step_cr_fail(context: Context, msg: str) -> None:
|
|
context.sv_result = SemanticCheckResult(
|
|
passed=False,
|
|
message=msg,
|
|
data={"detail": "x"},
|
|
severity=SemanticValidationSeverity.ERROR,
|
|
)
|
|
|
|
|
|
@then("the semantic check result passed is true")
|
|
def step_r_true(context: Context) -> None:
|
|
assert context.sv_result.passed is True
|
|
|
|
|
|
@then("the semantic check result passed is false")
|
|
def step_r_false(context: Context) -> None:
|
|
assert context.sv_result.passed is False
|
|
|
|
|
|
@then('the semantic check result message is "{msg}"')
|
|
def step_r_msg(context: Context, msg: str) -> None:
|
|
assert context.sv_result.message == msg
|
|
|
|
|
|
@then("the semantic check result data is none")
|
|
def step_r_none(context: Context) -> None:
|
|
assert context.sv_result.data is None
|
|
|
|
|
|
@then('the semantic check result has data key "{key}"')
|
|
def step_r_dk(context: Context, key: str) -> None:
|
|
assert context.sv_result.data is not None and key in context.sv_result.data
|
|
|
|
|
|
@then('the semantic check result message contains "{text}"')
|
|
def step_r_mc(context: Context, text: str) -> None:
|
|
assert text in context.sv_result.message
|
|
|
|
|
|
@then('the semantic check result severity is "{sev}"')
|
|
def step_r_sev(context: Context, sev: str) -> None:
|
|
assert context.sv_result.severity.value == sev
|
|
|
|
|
|
# -- Source code given steps -------------------------------------------------
|
|
@given("valid Python source code")
|
|
def step_src_valid(context: Context) -> None:
|
|
context.sv_source = VALID_PY
|
|
|
|
|
|
@given("Python source code with a syntax error")
|
|
def step_src_syn(context: Context) -> None:
|
|
context.sv_source = SYNTAX_ERR_PY
|
|
|
|
|
|
@given("Python source with standard imports")
|
|
def step_src_std(context: Context) -> None:
|
|
context.sv_source = STD_IMPORT_PY
|
|
|
|
|
|
@given("Python source with suspicious private imports")
|
|
def step_src_sus(context: Context) -> None:
|
|
context.sv_source = SUSPICIOUS_IMPORT_PY
|
|
|
|
|
|
@given("Python source with all names defined")
|
|
def step_src_def(context: Context) -> None:
|
|
context.sv_source = ALL_DEFINED_PY
|
|
|
|
|
|
@given("Python source with undefined references")
|
|
def step_src_uref(context: Context) -> None:
|
|
context.sv_source = UNDEF_REF_PY
|
|
|
|
|
|
@given("Python source with unique relative imports")
|
|
def step_src_urel(context: Context) -> None:
|
|
context.sv_source = UNIQUE_REL_PY
|
|
|
|
|
|
@given("Python source with duplicate relative imports")
|
|
def step_src_drel(context: Context) -> None:
|
|
context.sv_source = DUP_REL_PY
|
|
|
|
|
|
@given("Python source without API misuse")
|
|
def step_src_clean(context: Context) -> None:
|
|
context.sv_source = CLEAN_PY
|
|
|
|
|
|
@given("Python source using eval")
|
|
def step_src_eval(context: Context) -> None:
|
|
context.sv_source = EVAL_PY
|
|
|
|
|
|
@given("Python source with well-scoped functions")
|
|
def step_src_scoped(context: Context) -> None:
|
|
context.sv_source = SCOPED_PY
|
|
|
|
|
|
@given("Python source with undefined symbols in functions")
|
|
def step_src_usym(context: Context) -> None:
|
|
context.sv_source = UNDEF_SYM_PY
|
|
|
|
|
|
@given("Python source with function-local variables")
|
|
def step_src_func_local(context: Context) -> None:
|
|
context.sv_source = FUNC_LOCAL_PY
|
|
|
|
|
|
@given("Python source with a class method using undefined symbol")
|
|
def step_src_cls_method(context: Context) -> None:
|
|
context.sv_source = CLASS_METHOD_PY
|
|
|
|
|
|
@given("Python source with nested functions")
|
|
def step_src_nested(context: Context) -> None:
|
|
context.sv_source = NESTED_FUNC_PY
|
|
|
|
|
|
@given("Python source with annotated assignment")
|
|
def step_src_annassign(context: Context) -> None:
|
|
context.sv_source = ANNASSIGN_PY
|
|
|
|
|
|
@given("Python source with with-statement variable")
|
|
def step_src_with_stmt(context: Context) -> None:
|
|
context.sv_source = WITH_STMT_PY
|
|
|
|
|
|
@given("Python source using exec")
|
|
def step_src_exec(context: Context) -> None:
|
|
context.sv_source = EXEC_PY
|
|
|
|
|
|
@given("Python source using os.popen")
|
|
def step_src_os_popen(context: Context) -> None:
|
|
context.sv_source = OS_POPEN_PY
|
|
|
|
|
|
@given("Python source using subprocess.run")
|
|
def step_src_subprocess_run(context: Context) -> None:
|
|
context.sv_source = SUBPROCESS_RUN_PY
|
|
|
|
|
|
@given("Python source with comprehension variables")
|
|
def step_src_comprehension(context: Context) -> None:
|
|
context.sv_source = COMPREHENSION_PY
|
|
|
|
|
|
@given("empty Python source code")
|
|
def step_src_empty(context: Context) -> None:
|
|
context.sv_source = EMPTY_PY
|
|
|
|
|
|
@given("Python source with eval in a string literal")
|
|
def step_src_eval_str(context: Context) -> None:
|
|
context.sv_source = 'description = "Do not use eval() in production"\n'
|
|
|
|
|
|
@given("a non-Python filename")
|
|
def step_non_py_filename(context: Context) -> None:
|
|
context.sv_filename = NON_PY_FILENAME
|
|
|
|
|
|
# -- Rule execution ----------------------------------------------------------
|
|
@when("the syntax check rule runs")
|
|
def step_run_syn(context: Context) -> None:
|
|
context.sv_result = SyntaxCheckRule().check(context.sv_source, context.sv_filename)
|
|
|
|
|
|
@when("the missing import rule runs")
|
|
def step_run_imp(context: Context) -> None:
|
|
context.sv_result = MissingImportRule().check(
|
|
context.sv_source, context.sv_filename
|
|
)
|
|
|
|
|
|
@when("the broken reference rule runs")
|
|
def step_run_ref(context: Context) -> None:
|
|
context.sv_result = BrokenReferenceRule().check(
|
|
context.sv_source, context.sv_filename
|
|
)
|
|
|
|
|
|
@when("the dependency cycle rule runs")
|
|
def step_run_cyc(context: Context) -> None:
|
|
context.sv_result = DependencyCycleRule().check(
|
|
context.sv_source, context.sv_filename
|
|
)
|
|
|
|
|
|
@when("the API misuse rule runs")
|
|
def step_run_api(context: Context) -> None:
|
|
context.sv_result = APIMisuseRule().check(context.sv_source, context.sv_filename)
|
|
|
|
|
|
@when("the missing symbol rule runs")
|
|
def step_run_sym(context: Context) -> None:
|
|
context.sv_result = MissingSymbolRule().check(
|
|
context.sv_source, context.sv_filename
|
|
)
|
|
|
|
|
|
@when("the duplicate import rule runs")
|
|
def step_run_dup(context: Context) -> None:
|
|
context.sv_result = DuplicateImportRule().check(
|
|
context.sv_source, context.sv_filename
|
|
)
|
|
|
|
|
|
# -- Registry ----------------------------------------------------------------
|
|
@given("a default semantic rule registry")
|
|
def step_reg(context: Context) -> None:
|
|
context.sv_registry = create_default_registry()
|
|
|
|
|
|
@then("the registry has {count:d} rules")
|
|
def step_reg_cnt(context: Context, count: int) -> None:
|
|
assert len(context.sv_registry) == count
|
|
|
|
|
|
@then('the registry contains rule "{name}"')
|
|
def step_reg_has(context: Context, name: str) -> None:
|
|
assert context.sv_registry.get(name) is not None
|
|
|
|
|
|
@then('the registry returns None for rule "{name}"')
|
|
def step_reg_none(context: Context, name: str) -> None:
|
|
assert context.sv_registry.get(name) is None
|
|
|
|
|
|
@when('the rule "{name}" is removed from the registry')
|
|
def step_reg_rm(context: Context, name: str) -> None:
|
|
context.sv_registry.remove(name)
|
|
|
|
|
|
@then('removing rule "{name}" returns false')
|
|
def step_reg_rm_f(context: Context, name: str) -> None:
|
|
assert context.sv_registry.remove(name) is False
|
|
|
|
|
|
# -- Cache -------------------------------------------------------------------
|
|
@given("a semantic validation cache")
|
|
def step_cache(context: Context) -> None:
|
|
context.sv_cache = SemanticValidationCache()
|
|
|
|
|
|
@given("a semantic validation cache with max size {n:d}")
|
|
def step_cache_max(context: Context, n: int) -> None:
|
|
context.sv_cache = SemanticValidationCache(max_size=n)
|
|
|
|
|
|
@given('a cached semantic result for rule "{rule}" and hash "{h}"')
|
|
def step_cache_put(context: Context, rule: str, h: str) -> None:
|
|
context.sv_cache.put(rule, h, SemanticCheckResult(passed=True, message="cached"))
|
|
|
|
|
|
@then('the cache returns None for rule "{rule}" and hash "{h}"')
|
|
def step_cache_miss(context: Context, rule: str, h: str) -> None:
|
|
assert context.sv_cache.get(rule, h) is None
|
|
|
|
|
|
@then('the cache returns the result for rule "{rule}" and hash "{h}"')
|
|
def step_cache_hit(context: Context, rule: str, h: str) -> None:
|
|
r = context.sv_cache.get(rule, h)
|
|
assert r is not None and r.message == "cached"
|
|
|
|
|
|
@when('the cache entry for rule "{rule}" and hash "{h}" is invalidated')
|
|
def step_cache_inv(context: Context, rule: str, h: str) -> None:
|
|
context.sv_cache.invalidate(rule, h)
|
|
|
|
|
|
@when("the semantic cache is cleared")
|
|
def step_cache_clr(context: Context) -> None:
|
|
context.sv_cache.clear()
|
|
|
|
|
|
@then("the semantic cache has {count:d} entries")
|
|
def step_cache_cnt(context: Context, count: int) -> None:
|
|
assert len(context.sv_cache) == count
|
|
|
|
|
|
@then('the semantic cache hash for "{text}" is consistent')
|
|
def step_cache_hash(context: Context, text: str) -> None:
|
|
h1 = SemanticValidationCache.compute_hash(text)
|
|
assert h1 == SemanticValidationCache.compute_hash(text) and len(h1) == 64
|
|
|
|
|
|
# -- Severity mapping --------------------------------------------------------
|
|
@then('semantic severity "{sev}" maps to validation mode "{mode}"')
|
|
def step_sev_map(context: Context, sev: str, mode: str) -> None:
|
|
assert map_severity_to_mode(SemanticValidationSeverity(sev)).value == mode
|
|
|
|
|
|
@then('resolve_severity for "{rule}" returns "{expected}"')
|
|
def step_resolve(context: Context, rule: str, expected: str) -> None:
|
|
assert resolve_severity(rule).value == expected
|
|
|
|
|
|
@given('a custom severity mapping with "{rule}" as "{sev}"')
|
|
def step_cust_map(context: Context, rule: str, sev: str) -> None:
|
|
context.sv_custom_mapping = {rule: sev}
|
|
|
|
|
|
@then('resolve_severity with custom mapping for "{rule}" returns "{expected}"')
|
|
def step_resolve_c(context: Context, rule: str, expected: str) -> None:
|
|
assert resolve_severity(rule, context.sv_custom_mapping).value == expected
|
|
|
|
|
|
# -- Service -----------------------------------------------------------------
|
|
@given("a semantic validation service")
|
|
def step_svc(context: Context) -> None:
|
|
context.sv_service = SemanticValidationService()
|
|
|
|
|
|
@given("a semantic validation service with semantic disabled")
|
|
def step_svc_off(context: Context) -> None:
|
|
context.sv_service = SemanticValidationService(config={CONFIG_KEY_ENABLED: False})
|
|
|
|
|
|
@given("a semantic validation service with python disabled")
|
|
def step_svc_pyoff(context: Context) -> None:
|
|
context.sv_service = SemanticValidationService(
|
|
config={CONFIG_KEY_PYTHON_ENABLED: False},
|
|
)
|
|
|
|
|
|
@when("the service checks the file")
|
|
def step_svc_chk(context: Context) -> None:
|
|
context.sv_results = context.sv_service.check_file(
|
|
context.sv_source,
|
|
context.sv_filename,
|
|
)
|
|
|
|
|
|
@when("the service checks the python file")
|
|
def step_svc_chk_py(context: Context) -> None:
|
|
context.sv_results = context.sv_service.check_file(context.sv_source, "module.py")
|
|
|
|
|
|
@when('the service checks the file with rules "{rules}"')
|
|
def step_svc_chk_r(context: Context, rules: str) -> None:
|
|
context.sv_results = context.sv_service.check_file(
|
|
context.sv_source,
|
|
context.sv_filename,
|
|
rule_names=[r.strip() for r in rules.split(",")],
|
|
)
|
|
|
|
|
|
@when("the service checks the file twice")
|
|
def step_svc_chk2(context: Context) -> None:
|
|
context.sv_service.check_file(context.sv_source, context.sv_filename)
|
|
context.sv_results = context.sv_service.check_file(
|
|
context.sv_source,
|
|
context.sv_filename,
|
|
)
|
|
|
|
|
|
@when("the service checks the non-python file")
|
|
def step_svc_chk_nonpy(context: Context) -> None:
|
|
context.sv_filename = NON_PY_FILENAME
|
|
context.sv_results = context.sv_service.check_file(
|
|
context.sv_source,
|
|
context.sv_filename,
|
|
)
|
|
|
|
|
|
@then("all semantic check results passed")
|
|
def step_all_pass(context: Context) -> None:
|
|
assert all(r.passed for r in context.sv_results)
|
|
|
|
|
|
@then("at least one semantic check result failed")
|
|
def step_some_fail(context: Context) -> None:
|
|
assert any(not r.passed for r in context.sv_results)
|
|
|
|
|
|
@then('a failing result has message containing "{text}"')
|
|
def step_fail_msg(context: Context, text: str) -> None:
|
|
assert any(text in r.message for r in context.sv_results if not r.passed)
|
|
|
|
|
|
@then("only {count:d} semantic check result is returned")
|
|
def step_cnt1(context: Context, count: int) -> None:
|
|
assert len(context.sv_results) == count
|
|
|
|
|
|
@then("{count:d} semantic check results are returned")
|
|
def step_cnt(context: Context, count: int) -> None:
|
|
assert len(context.sv_results) == count
|
|
|
|
|
|
@then("the cache has entries")
|
|
def step_cache_has(context: Context) -> None:
|
|
assert len(context.sv_service.cache) > 0
|
|
|
|
|
|
# -- Pipeline integration ---------------------------------------------------
|
|
@when("the service returns pipeline results")
|
|
def step_pipe(context: Context) -> None:
|
|
context.sv_pipeline_results = context.sv_service.as_pipeline_results(
|
|
context.sv_source,
|
|
context.sv_filename,
|
|
)
|
|
|
|
|
|
@then('each pipeline result has keys "passed" and "message" and "data"')
|
|
def step_pipe_k(context: Context) -> None:
|
|
for pr in context.sv_pipeline_results:
|
|
assert "passed" in pr and "message" in pr and "data" in pr
|
|
|
|
|
|
@then('each pipeline result has key "{key}"')
|
|
def step_pipe_k1(context: Context, key: str) -> None:
|
|
for pr in context.sv_pipeline_results:
|
|
assert key in pr
|
|
|
|
|
|
@when("the result is normalised")
|
|
def step_norm(context: Context) -> None:
|
|
context.sv_normalised = SemanticValidationService().normalise_output(
|
|
context.sv_result
|
|
)
|
|
|
|
|
|
@then('the normalised output has keys "passed" and "message" and "data"')
|
|
def step_norm_k(context: Context) -> None:
|
|
n = context.sv_normalised
|
|
assert "passed" in n and "message" in n and "data" in n
|
|
|
|
|
|
# -- Config keys -------------------------------------------------------------
|
|
@then('config key "{key}" exists')
|
|
def step_cfg(context: Context, key: str) -> None:
|
|
assert key in DEFAULT_CONFIG
|
|
|
|
|
|
# -- Required vs informational -----------------------------------------------
|
|
@then('a pipeline result with severity "{sev}" has mode "{mode}"')
|
|
def step_pipe_mode(context: Context, sev: str, mode: str) -> None:
|
|
matching = [r for r in context.sv_pipeline_results if r["severity"] == sev]
|
|
assert len(matching) > 0
|
|
assert all(r["mode"] == mode for r in matching)
|
|
|
|
|
|
# -- Safe initialisation and cleanup -----------------------------------------
|
|
@then("creating a semantic validation cache with max size 0 raises ValueError")
|
|
def step_cache_invalid_size(context: Context) -> None:
|
|
raised = False
|
|
try:
|
|
SemanticValidationCache(max_size=0)
|
|
except ValueError:
|
|
raised = True
|
|
assert raised, "Expected ValueError for max_size=0"
|
|
|
|
|
|
@when(
|
|
'the cache entry for rule "{rule}" and hash "{h}" is overwritten '
|
|
'with message "{msg}"'
|
|
)
|
|
def step_cache_overwrite(context: Context, rule: str, h: str, msg: str) -> None:
|
|
context.sv_cache.put(rule, h, SemanticCheckResult(passed=True, message=msg))
|
|
|
|
|
|
@then(
|
|
'the cache returns the updated result for rule "{rule}" and hash "{h}" '
|
|
'with message "{msg}"'
|
|
)
|
|
def step_cache_updated(context: Context, rule: str, h: str, msg: str) -> None:
|
|
r = context.sv_cache.get(rule, h)
|
|
assert r is not None and r.message == msg
|
|
|
|
|
|
@then("the registry list_rules returns sorted rule names")
|
|
def step_reg_list(context: Context) -> None:
|
|
names = context.sv_registry.list_rules()
|
|
assert names == sorted(names)
|
|
assert len(names) == len(context.sv_registry)
|
|
|
|
|
|
@then("the service registry property returns a registry with {count:d} rules")
|
|
def step_svc_reg(context: Context, count: int) -> None:
|
|
assert len(context.sv_service.registry) == count
|
|
|
|
|
|
@given("a semantic validation service with non-dict severity mapping")
|
|
def step_svc_nondict_sev(context: Context) -> None:
|
|
context.sv_service = SemanticValidationService(
|
|
config={CONFIG_KEY_SEVERITY_MAPPING: "not-a-dict"},
|
|
)
|
|
|
|
|
|
# -- MissingImportRule from-import form --------------------------------------
|
|
@given("Python source with from-import of private module")
|
|
def step_src_from_priv(context: Context) -> None:
|
|
context.sv_source = FROM_PRIVATE_IMPORT_PY
|
|
|
|
|
|
# -- BrokenReferenceRule advanced scope patterns -----------------------------
|
|
@given("Python source with varargs and kwargs function")
|
|
def step_src_varargs(context: Context) -> None:
|
|
context.sv_source = VARARGS_KWARGS_PY
|
|
|
|
|
|
@given("Python source with from-import bindings")
|
|
def step_src_from_import(context: Context) -> None:
|
|
context.sv_source = FROM_IMPORT_BINDINGS_PY
|
|
|
|
|
|
@given("Python source with except handler variable")
|
|
def step_src_except(context: Context) -> None:
|
|
context.sv_source = EXCEPT_HANDLER_PY
|
|
|
|
|
|
@given("Python source with tuple and starred unpacking")
|
|
def step_src_tuple_star(context: Context) -> None:
|
|
context.sv_source = TUPLE_STARRED_PY
|
|
|
|
|
|
@given("Python source with with-statement without as clause")
|
|
def step_src_with_no_as(context: Context) -> None:
|
|
context.sv_source = WITH_NO_AS_PY
|
|
|
|
|
|
# -- APIMisuseRule safe attribute on flagged module --------------------------
|
|
@given("Python source calling a safe method on a flagged module")
|
|
def step_src_safe_flagged(context: Context) -> None:
|
|
context.sv_source = SAFE_FLAGGED_MODULE_PY
|
|
|
|
|
|
# -- MissingSymbolRule function-local binding patterns -----------------------
|
|
@given("Python source with function using varargs kwonly and kwargs")
|
|
def step_src_func_all_params(context: Context) -> None:
|
|
context.sv_source = FUNC_ALL_PARAMS_PY
|
|
|
|
|
|
@given("Python source with function containing for-loop variable")
|
|
def step_src_func_for(context: Context) -> None:
|
|
context.sv_source = FUNC_FOR_LOOP_PY
|
|
|
|
|
|
@given("Python source with function containing with-statement variable")
|
|
def step_src_func_with(context: Context) -> None:
|
|
context.sv_source = FUNC_WITH_STMT_PY
|
|
|
|
|
|
@given("Python source with function containing except handler variable")
|
|
def step_src_func_except(context: Context) -> None:
|
|
context.sv_source = FUNC_EXCEPT_PY
|
|
|
|
|
|
@given("Python source with function containing local import")
|
|
def step_src_func_import(context: Context) -> None:
|
|
context.sv_source = FUNC_LOCAL_IMPORT_PY
|
|
|
|
|
|
@given("Python source with function containing local from-import")
|
|
def step_src_func_from_import(context: Context) -> None:
|
|
context.sv_source = FUNC_LOCAL_FROM_IMPORT_PY
|