#!/usr/bin/env python3 """Self-test for exchange_lint.py. Builds two throwaway Ignition projects in a temp directory -- one that follows the Exchange Resources Style Guide and one that breaks a known rule per check -- then asserts the checker stays silent on the first and fires the expected rule id on the second. This is what keeps the rules honest: a rule that cannot fail and a rule that always fires both show up here. Run directly (python3 test_exchange_lint.py) or under pytest. """ import json import os import shutil import sys import tempfile import unittest sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import exchange_lint as el # noqa: E402 RESOURCE = "SampleResource" PERSP = el.PERSPECTIVE # ------------------------------------------------------------------ fixtures def write(path, text): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as f: f.write(text) def write_json(path, obj): write(path, json.dumps(obj, indent=2)) RESOURCE_JSON = {"scope": "G", "version": 1, "restricted": False, "overridable": True, "files": ["view.json"], "attributes": {}} CLEAN_CODE = '''"""Sample library module.""" def formatCount(rawCount): """Return rawCount rendered for display.""" # pad small values so the column stays aligned displayValue = "%d" % rawCount return displayValue ''' DIRTY_CODE = '''def format_count(raw_count): #no space after hash display_value = "%d" % raw_count return display_value ''' def build_project(root, name, clean=True): """Create a minimal Perspective project; clean=False breaks one rule per check.""" proj = os.path.join(root, name) ns_upper = "Exchange/%s" % RESOURCE if clean else "MyCompany" ns_lower = "exchange/sample-resource" if clean else "MyCompany" lib_pkg = "exchange/sampleresource" if clean else "MyCompany" write_json(os.path.join(proj, "project.json"), { "title": "Sample Resource" if clean else "", "description": "A sample Exchange resource." if clean else "", "enabled": True, "inheritable": False, "parent": "", }) view_dir = os.path.join(proj, PERSP, "views", *ns_upper.split("/"), "Dashboard") write_json(os.path.join(view_dir, "resource.json"), RESOURCE_JSON) write_json(os.path.join(view_dir, "view.json"), { "params": {"startMs": 0} if clean else {"start_ms": 0}, "custom": {}, "root": { "type": "ia.container.flex", "meta": {"name": "root"}, "props": {"style": {"classes": "%s/dashboard/card" % ns_lower if clean else "%s/Card" % ns_lower}}, "children": [{ "type": "ia.display.label", "meta": {"name": "TitleLabel" if clean else "titleLabel"}, "custom": {"labelText": ""} if clean else {"label_text": ""}, "props": {}, "scripts": {"messageHandlers": [{ "messageType": ("exchange.sampleResource.refresh" if clean else "refresh")}]}, }], }, }) style_dir = os.path.join(proj, PERSP, "style-classes", *ns_lower.split("/"), *(("dashboard", "card") if clean else ("Card",))) write_json(os.path.join(style_dir, "resource.json"), dict(RESOURCE_JSON, files=["style.json"])) write_json(os.path.join(style_dir, "style.json"), {"style": {}}) write_json(os.path.join(proj, PERSP, "page-config", "config.json"), {"pages": {"/sample-resource" if clean else "/": { "viewPath": "%s/Dashboard" % ns_upper}}, "sharedDocks": {}}) write_json(os.path.join(proj, PERSP, "session-props", "props.json"), {"custom": {} if clean else {"MyCompany": {"selectedTab": 0}}, "props": {}}) code_dir = os.path.join(proj, "ignition", "script-python", *lib_pkg.split("/"), "fmt") write_json(os.path.join(code_dir, "resource.json"), dict(RESOURCE_JSON, scope="A", files=["code.py"])) write(os.path.join(code_dir, "code.py"), CLEAN_CODE if clean else DIRTY_CODE) if clean: write(os.path.join(proj, "README.md"), "# Sample Resource\n\nInstall steps.\n") else: vis_ctags = os.path.join(proj, el.VISION, "client-tags") write_json(os.path.join(vis_ctags, "resource.json"), dict(RESOURCE_JSON, files=["data.bin"])) write(os.path.join(vis_ctags, "data.bin"), "x") return proj SQL_CLEAN = """ CREATE TABLE ex_sr_contacts ( id INT NOT NULL AUTO_INCREMENT, contact_name VARCHAR(64), PRIMARY KEY (id) ); CREATE TABLE ex_sr_calls ( id INT NOT NULL AUTO_INCREMENT, ex_sr_contacts_id INT, PRIMARY KEY (id), FOREIGN KEY (ex_sr_contacts_id) REFERENCES ex_sr_contacts(id) ); """ SQL_DIRTY = """ CREATE TABLE Contacts ( contactId INT NOT NULL AUTO_INCREMENT, ContactName VARCHAR(64), PRIMARY KEY (contactId) ); CREATE TABLE ex_sr_calls ( id INT NOT NULL AUTO_INCREMENT, contact INT, PRIMARY KEY (id), FOREIGN KEY (contact) REFERENCES Contacts(contactId) ); """ TAGS_CLEAN = {"tags": [{"name": "Exchange", "tagType": "Folder", "tags": [ {"name": RESOURCE, "tagType": "Folder", "tags": [ {"name": "AlarmCount", "tagType": "AtomicTag", "valueSource": "memory"}]}]}]} TAGS_DIRTY = {"tags": [{"name": "my_company", "tagType": "Folder", "tags": [ {"name": "alarm_count", "tagType": "AtomicTag", "valueSource": "memory"}]}]} # --------------------------------------------------------------------- tests class ExchangeLintTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.tmp = tempfile.mkdtemp(prefix="exlint_test_") cls.clean = build_project(cls.tmp, "sample-resource", clean=True) cls.dirty = build_project(cls.tmp, "SampleResource_v2", clean=False) @classmethod def tearDownClass(cls): shutil.rmtree(cls.tmp, ignore_errors=True) def run_lint(self, proj, **kw): kw.setdefault("tags", []) kw.setdefault("sql", []) kw.setdefault("resource", RESOURCE) rep, _name, _res = el.lint_project(proj, **kw) return rep def rules(self, rep): return {f[1] for f in rep.findings} # -- the clean project must be silent --------------------------------- def test_clean_project_has_no_findings(self): rep = self.run_lint(self.clean) self.assertEqual(rep.findings, [], "conforming project produced findings:\n" + "\n".join("%s %s %s: %s" % f for f in rep.findings)) def test_clean_project_exits_zero(self): rc = el.main([self.clean, "--resource", RESOURCE, "--tags", "--sql"]) self.assertEqual(rc, 0) # -- the dirty project must fire each rule ---------------------------- def test_dirty_project_fires_expected_rules(self): got = self.rules(self.run_lint(self.dirty)) for rule in ("E-PROJ-NAME", "E-PROJ-TITLE", "E-PROJ-DESC", "E-NS-ROOT", "E-COMP-NAME", "E-PROP-NAME", "E-PAGE-ROOT", "E-STYLE-NAME", "E-MSG-NAME", "E-STORE-SESS", "E-STORE-CTAG", "E-PY-TABS", "E-PY-FUNC", "E-PY-VAR", "E-PY-DOC", "E-PY-COMMENT", "E-UP-DOCS"): self.assertIn(rule, got, "expected %s to fire" % rule) def test_dirty_project_exits_nonzero(self): rc = el.main([self.dirty, "--resource", RESOURCE, "--tags", "--sql"]) self.assertEqual(rc, 1) # -- severity model --------------------------------------------------- def test_namespace_and_storage_are_failures(self): rep = self.run_lint(self.dirty) fails = {f[1] for f in rep.fails} for rule in ("E-NS-ROOT", "E-STORE-SESS", "E-STORE-CTAG", "E-PROJ-NAME"): self.assertIn(rule, fails, "%s should be a FAIL, not a warning" % rule) def test_naming_rules_are_warnings_by_default(self): rep = self.run_lint(self.dirty) warns = {f[1] for f in rep.warns} for rule in ("E-COMP-NAME", "E-PY-FUNC", "E-PY-TABS"): self.assertIn(rule, warns, "%s should warn by default" % rule) def test_strict_promotes_warnings(self): rep = self.run_lint(self.dirty, strict=True) self.assertEqual(rep.warns, [], "--strict left warnings behind") self.assertIn("E-COMP-NAME", {f[1] for f in rep.fails}) def test_ignore_suppresses_a_rule(self): rep = self.run_lint(self.dirty, ignore={"E-COMP-NAME"}) self.assertNotIn("E-COMP-NAME", self.rules(rep)) # -- database and tag checks ----------------------------------------- def test_sql_clean_and_dirty(self): clean_sql = os.path.join(self.tmp, "clean.sql") dirty_sql = os.path.join(self.tmp, "dirty.sql") write(clean_sql, SQL_CLEAN) write(dirty_sql, SQL_DIRTY) rep = el.Report() el.check_database(rep, [clean_sql], RESOURCE) self.assertEqual(rep.findings, [], "conforming schema produced findings: %s" % rep.findings) rep = el.Report() el.check_database(rep, [dirty_sql], RESOURCE) got = {f[1] for f in rep.findings} for rule in ("E-DB-TABLE", "E-DB-COL", "E-DB-ID", "E-DB-FK"): self.assertIn(rule, got, "expected %s to fire" % rule) def test_tags_clean_and_dirty(self): clean_tags = os.path.join(self.tmp, "clean-tags.json") dirty_tags = os.path.join(self.tmp, "dirty-tags.json") write_json(clean_tags, TAGS_CLEAN) write_json(dirty_tags, TAGS_DIRTY) rep = el.Report() el.check_tags(rep, [clean_tags], RESOURCE) self.assertEqual(rep.findings, [], "conforming tag export produced findings: %s" % rep.findings) rep = el.Report() el.check_tags(rep, [dirty_tags], RESOURCE) got = {f[1] for f in rep.findings} self.assertIn("E-NS-ROOT", got) self.assertIn("E-TAG-NAME", got) # -- discovery -------------------------------------------------------- def test_discovers_both_projects_from_a_containing_folder(self): found = el.discover_projects(self.tmp) self.assertEqual(sorted(os.path.basename(p) for p in found), ["SampleResource_v2", "sample-resource"]) def test_project_folder_resolves_to_itself(self): self.assertEqual(el.discover_projects(self.clean), [self.clean]) def test_zip_export_is_unpacked(self): archive = shutil.make_archive( os.path.join(self.tmp, "export"), "zip", root_dir=os.path.dirname(self.clean), base_dir=os.path.basename(self.clean)) found = el.discover_projects(archive) self.assertEqual(len(found), 1) self.assertTrue(os.path.exists(os.path.join(found[0], "project.json"))) def test_resource_name_inferred_from_namespace(self): self.assertEqual(el.infer_resource(self.clean, "sample-resource"), RESOURCE) # -- name predicates -------------------------------------------------- def test_name_predicates(self): self.assertTrue(el.is_pascal("AreaCard")) self.assertFalse(el.is_pascal("areaCard")) self.assertTrue(el.is_camel("startMs")) self.assertFalse(el.is_camel("start_ms")) self.assertTrue(el.is_kebab("bad-actors")) self.assertFalse(el.is_kebab("BadActors")) self.assertTrue(el.is_title("Alarm Dashboard")) self.assertTrue(el.is_title("Table of Contents")) self.assertFalse(el.is_title("alarm dashboard")) self.assertTrue(el.is_title_or_pascal("AlarmDashboard")) def test_rule_table_ids_are_unique_and_referenced(self): ids = [r[0] for r in el.RULES] self.assertEqual(len(ids), len(set(ids)), "duplicate rule id in RULES") with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "exchange_lint.py")) as f: source = f.read() for rid in ids: self.assertGreaterEqual(source.count('"%s"' % rid), 2, "%s is declared but never raised" % rid) if __name__ == "__main__": unittest.main(verbosity=2)