Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311#!/usr/bin/env python3"""Guard the appview content-type normalizer's single production ownership."""
from __future__ import annotations
import argparseimport reimport tempfilefrom pathlib import Path
CALLER_FILES = ("proxy.rs", "writes.rs", "drafts.rs")IMPORT = "use super::content_type::content_type;"CALL = "content_type(&headers)"DEFINITION = "pub(super) fn content_type(headers: &HeaderMap) -> String"CONTENT_TYPE_FUNCTION = re.compile(r"\bfn\s+content_type\s*\(")CFG_TEST = re.compile(r"#\[\s*cfg\s*\(\s*test\s*\)\s*\]")TEST_ITEM = re.compile( r"\s*(?:(?:#\[[^]]*\]\s*)*)" r"(?:(?:pub(?:\([^)]*\))?\s+)?(?:async\s+)?)" r"(?P<kind>mod|fn)\b")
def _mask_non_code(text: str) -> str: """Replace comments and literals with spaces while preserving positions.""" masked = list(text) index = 0 length = len(text) while index < length: if text.startswith("//", index): end = text.find("\n", index) end = length if end == -1 else end for position in range(index, end): masked[position] = " " index = end continue if text.startswith("/*", index): depth = 1 end = index + 2 while end < length and depth: if text.startswith("/*", end): depth += 1 end += 2 elif text.startswith("*/", end): depth -= 1 end += 2 else: end += 1 for position in range(index, end): if text[position] != "\n": masked[position] = " " index = end continue if text[index] == "r": hash_end = index + 1 while hash_end < length and text[hash_end] == "#": hash_end += 1 if hash_end < length and text[hash_end] == '"': terminator = '"' + ("#" * (hash_end - index - 1)) end = text.find(terminator, hash_end + 1) end = length if end == -1 else end + len(terminator) for position in range(index, end): if text[position] != "\n": masked[position] = " " index = end continue if text[index] == '"': end = index + 1 while end < length: if text[end] == "\\": end += 2 elif text[end] == '"': end += 1 break else: end += 1 for position in range(index, min(end, length)): if text[position] != "\n": masked[position] = " " index = end continue index += 1 return "".join(masked)
def _matching_brace(text: str, opening: int) -> int | None: depth = 0 for index in range(opening, len(text)): if text[index] == "{": depth += 1 elif text[index] == "}": depth -= 1 if depth == 0: return index + 1 return None
def _matching_delimiter(text: str, opening: int) -> int | None: closing_for = {"{": "}", "(": ")", "[": "]"} expected = [closing_for.get(text[opening])] if expected[0] is None: return None for index in range(opening + 1, len(text)): if text[index] in closing_for: expected.append(closing_for[text[index]]) elif text[index] == expected[-1]: expected.pop() if not expected: return index + 1 return None
def _mask_macro_bodies(masked: str) -> str: """Mask macro token trees while preserving source positions.""" source = masked ranges: list[tuple[int, int]] = [] for marker in re.finditer(r"\bmacro_rules\s*!", source): opening = source.find("{", marker.end()) if opening != -1: closing = _matching_delimiter(source, opening) if closing is not None: ranges.append((marker.start(), closing)) for marker in re.finditer(r"(?:\b\w+::)*\w+!", source): opening = marker.end() while opening < len(source) and source[opening].isspace(): opening += 1 if opening >= len(source) or source[opening] not in "{([": continue closing = _matching_delimiter(source, opening) if closing is not None: ranges.append((marker.start(), closing)) masked = list(masked) for start, end in ranges: for position in range(start, end): if masked[position] != "\n": masked[position] = " " return "".join(masked)
def _cfg_test_spans(masked: str) -> list[tuple[int, int]]: spans: list[tuple[int, int]] = [] for marker in CFG_TEST.finditer(masked): item = TEST_ITEM.match(masked, marker.end()) if item is None: continue opening = masked.find("{", item.end()) if opening == -1: continue closing = _matching_brace(masked, opening) if closing is not None: spans.append((marker.start(), closing)) return spans
def _production_code(path: Path) -> tuple[str, list[tuple[int, int]]]: masked = _mask_non_code(path.read_text(encoding="utf-8")) masked = _mask_macro_bodies(masked) return masked, _cfg_test_spans(masked)
def _in_spans(position: int, spans: list[tuple[int, int]]) -> bool: return any(start <= position < end for start, end in spans)
def _production_content_type_definitions(path: Path) -> list[int]: masked, test_spans = _production_code(path) return [ match.start() for match in CONTENT_TYPE_FUNCTION.finditer(masked) if not _in_spans(match.start(), test_spans) ]
def _production_occurrences(path: Path, pattern: str) -> int: masked, test_spans = _production_code(path) return sum( 1 for match in re.finditer(re.escape(pattern), masked) if not _in_spans(match.start(), test_spans) )
def check_content_type_source(root: Path) -> list[str]: appview = root / "src" / "appview" helper = appview / "content_type.rs" errors: list[str] = []
if not helper.is_file(): errors.append("missing src/appview/content_type.rs") else: if _production_occurrences(helper, DEFINITION) != 1: errors.append("expected exactly one canonical content_type definition")
definitions: list[tuple[Path, int]] = [] for path in sorted(appview.rglob("*.rs")): if path.is_file(): definitions.extend( (path, position) for position in _production_content_type_definitions(path) ) if len(definitions) != 1 or definitions[0][0] != helper: for path, position in definitions: line = path.read_text(encoding="utf-8")[:position].count("\n") + 1 errors.append(f"{path.relative_to(root)}:{line}: duplicate content_type definition") if not definitions: errors.append("expected one production content_type definition under src/appview")
for filename in CALLER_FILES: path = appview / filename if not path.is_file(): errors.append(f"missing src/appview/{filename}") continue if _production_occurrences(path, IMPORT) != 1: errors.append(f"{filename}: expected one canonical helper import") if _production_occurrences(path, CALL) != 1: errors.append(f"{filename}: expected one canonical helper call")
return errors
def run_self_test() -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) appview = root / "src" / "appview" appview.mkdir(parents=True) (appview / "content_type.rs").write_text( f"{DEFINITION} {{}}\n", encoding="utf-8" ) for filename in CALLER_FILES: (appview / filename).write_text( f"{IMPORT}\nfn upload(headers: &HeaderMap) {{ let _ = {CALL}; }}\n", encoding="utf-8", ) errors = check_content_type_source(root) if errors: raise AssertionError(f"positive fixture failed: {errors}")
(appview / "writes.rs").write_text( f"{IMPORT}\nfn content_type(headers: &HeaderMap) -> String {{ String::new() }}\n" f"fn upload(headers: &HeaderMap) {{ let _ = {CALL}; }}\n", encoding="utf-8", ) errors = check_content_type_source(root) if not errors: raise AssertionError("negative fixture unexpectedly passed")
decoy_fixtures = { "comment": f"// {IMPORT}\n// {CALL}\n", "string": f'const IMPORT_DECOY: &str = "{IMPORT}";\n' f'const CALL_DECOY: &str = "{CALL}";\n', "macro": ( "macro_rules! decoy { () => { " + IMPORT + " fn upload(headers: &HeaderMap) { let _ = " + CALL + "; } } }\n" ), "cfg(test)": ( f"#[cfg(test)]\nmod tests {{\n {IMPORT}\n" f" fn upload(headers: &HeaderMap) {{ let _ = {CALL}; }}\n}}\n" ), } for label, fixture in decoy_fixtures.items(): (appview / "writes.rs").write_text(fixture, encoding="utf-8") errors = check_content_type_source(root) if not errors: raise AssertionError(f"{label} decoy unexpectedly passed")
(appview / "writes.rs").write_text( f"{IMPORT}\nfn upload(headers: &HeaderMap) {{ let _ = {CALL}; }}\n", encoding="utf-8", ) errors = check_content_type_source(root) if errors: raise AssertionError(f"test-only fixture failed: {errors}")
nested = appview / "nested" nested.mkdir() (nested / "helper.rs").write_text( "fn content_type(headers: &HeaderMap) -> String { String::new() }\n", encoding="utf-8", ) errors = check_content_type_source(root) if not errors: raise AssertionError("nested production duplicate unexpectedly passed")
print("PASS: check-content-type self-test")
def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: run_self_test() return 0
errors = check_content_type_source(Path(__file__).resolve().parents[1]) if errors: for error in errors: print(f"FAIL: {error}") return 1 print( "PASS: content_type_normalizer_has_one_definition_and_three_callers" ) return 0
if __name__ == "__main__": raise SystemExit(main())