Alex Hochheiden
c49fedbda1
Updated aiohttp v3.12.12 -> v3.12.13 Updated attrs v23.1.0 -> v23.2.0 Updated blessed v1.19.1 -> v1.21.0 Updated cbor2 v4.0.1 -> v4.1.2 Updated compare-locales v9.0.1 -> v9.0.4 Updated distro v1.8.0 -> v1.9.0 Updated fluent-migrate v0.13.2 -> v0.13.3 Updated importlib-metadata v6.0.0 -> v6.11.0 Updated jsmin v3.0.0 -> v3.0.1 Updated json-e v4.5.3 -> v4.8.0 Updated looseversion v1.0.1 -> v1.3.0 Updated mozilla-taskgraph v3.3.1 -> v3.3.2 Updated multidict v6.4.4 -> v6.5.0 Updated pathspec v0.9.0 -> v0.12.1 Updated pyasn1 v0.4.8 -> v0.6.1 Updated pyasn1-modules v0.2.8 -> v0.4.2 Updated pylru v1.0.9 -> v1.2.1 Updated python-hglib v2.4 -> v2.6.2 Updated redo v2.0.3 -> v2.0.4 Updated requests-unixsocket v0.2.0 -> v0.4.1 Updated responses v0.10.6 -> v0.25.7 Updated rsa v4.9 -> v4.9.1 Updated sentry-sdk v1.14.0 -> v1.45.1 Updated six v1.16.0 -> v1.17.0 Updated tomlkit v0.12.3 -> v0.13.3 Updated tqdm v4.66.3 -> v4.67.1 Updated typing-extensions v4.12.2 -> v4.14.0 Updated voluptuous v0.12.1 -> v0.15.2 Updated yamllint v1.23.0 -> v1.37.1 Differential Revision: https://phabricator.services.mozilla.com/D254096
324 lines
11 KiB
Python
324 lines
11 KiB
Python
import argparse
|
|
import ast
|
|
from itertools import zip_longest
|
|
|
|
from fluent.migrate import transforms
|
|
from fluent.migrate.errors import MigrationError
|
|
from fluent.migrate.helpers import transforms_from
|
|
from fluent.syntax.visitor import Visitor
|
|
from compare_locales import mozpath
|
|
|
|
|
|
class MigrateNotFoundException(Exception):
|
|
pass
|
|
|
|
|
|
class BadContextAPIException(Exception):
|
|
pass
|
|
|
|
|
|
def process_assign(node, context):
|
|
if isinstance(node.value, ast.Constant):
|
|
val = node.value.value
|
|
elif isinstance(node.value, ast.Name):
|
|
val = context.get(node.value.id)
|
|
elif isinstance(node.value, ast.Call):
|
|
val = node.value
|
|
else:
|
|
val = None
|
|
if val is None:
|
|
return
|
|
for target in node.targets:
|
|
if isinstance(target, ast.Name):
|
|
context[target.id] = val
|
|
|
|
|
|
class Validator:
|
|
"""Validate a migration recipe
|
|
|
|
Extract information from the migration recipe about which files to
|
|
migrate from, and which files to migrate to.
|
|
Also check for errors in the recipe, or bad API usage.
|
|
"""
|
|
|
|
@classmethod
|
|
def validate(cls, path, code=None):
|
|
if code is None:
|
|
with open(path) as fh:
|
|
code = fh.read()
|
|
validator = cls(code, path)
|
|
return validator.inspect()
|
|
|
|
def __init__(self, code, path):
|
|
self.ast = ast.parse(code, path)
|
|
|
|
def inspect(self):
|
|
migrate_func = None
|
|
global_assigns = {}
|
|
for top_level in ast.iter_child_nodes(self.ast):
|
|
if isinstance(top_level, ast.FunctionDef) and top_level.name == "migrate":
|
|
if migrate_func:
|
|
raise MigrateNotFoundException("Duplicate definition of migrate")
|
|
migrate_func = top_level
|
|
details = self.inspect_migrate(migrate_func, global_assigns)
|
|
if isinstance(top_level, ast.Assign):
|
|
process_assign(top_level, global_assigns)
|
|
if isinstance(top_level, (ast.Import, ast.ImportFrom)):
|
|
if "module" in top_level._fields:
|
|
module = top_level.module
|
|
else:
|
|
module = None
|
|
for alias in top_level.names:
|
|
asname = alias.asname or alias.name
|
|
dotted = alias.name
|
|
if module:
|
|
dotted = f"{module}.{dotted}"
|
|
global_assigns[asname] = dotted
|
|
if not migrate_func:
|
|
raise MigrateNotFoundException("migrate function not found")
|
|
return details
|
|
|
|
def inspect_migrate(self, migrate_func, global_assigns):
|
|
if len(migrate_func.args.args) != 1 or any(
|
|
getattr(migrate_func.args, arg_field)
|
|
for arg_field in migrate_func.args._fields
|
|
if arg_field != "args"
|
|
):
|
|
raise MigrateNotFoundException("migrate takes only one positional argument")
|
|
arg = migrate_func.args.args[0]
|
|
if isinstance(arg, ast.Name):
|
|
ctx_var = arg.id # python 2
|
|
else:
|
|
ctx_var = arg.arg # python 3
|
|
visitor = MigrateAnalyzer(ctx_var, global_assigns)
|
|
visitor.visit(migrate_func)
|
|
return {
|
|
"references": visitor.references,
|
|
"issues": visitor.issues,
|
|
}
|
|
|
|
|
|
def full_name(node, global_assigns):
|
|
leafs = []
|
|
while isinstance(node, ast.Attribute):
|
|
leafs.append(node.attr)
|
|
node = node.value
|
|
if isinstance(node, ast.Name):
|
|
leafs.append(global_assigns.get(node.id, node.id))
|
|
return ".".join(reversed(leafs))
|
|
|
|
|
|
PATH_TYPES = (str,) + (ast.Call,)
|
|
|
|
|
|
class MigrateAnalyzer(ast.NodeVisitor):
|
|
def __init__(self, ctx_var, global_assigns):
|
|
super().__init__()
|
|
self.ctx_var = ctx_var
|
|
self.global_assigns = global_assigns
|
|
self.depth = 0
|
|
self.issues = []
|
|
self.references = set()
|
|
|
|
def generic_visit(self, node):
|
|
self.depth += 1
|
|
super().generic_visit(node)
|
|
self.depth -= 1
|
|
|
|
def visit_Assign(self, node):
|
|
if self.depth == 1:
|
|
process_assign(node, self.global_assigns)
|
|
self.generic_visit(node)
|
|
|
|
def visit_Attribute(self, node):
|
|
if isinstance(node.value, ast.Name) and node.value.id == self.ctx_var:
|
|
if node.attr not in (
|
|
"add_transforms",
|
|
"locale",
|
|
):
|
|
raise BadContextAPIException(
|
|
"Unexpected attribute access on {}.{}".format(
|
|
self.ctx_var, node.attr
|
|
)
|
|
)
|
|
self.generic_visit(node)
|
|
|
|
def visit_Call(self, node):
|
|
if (
|
|
isinstance(node.func, ast.Attribute)
|
|
and isinstance(node.func.value, ast.Name)
|
|
and node.func.value.id == self.ctx_var
|
|
):
|
|
return self.call_ctx(node)
|
|
dotted = full_name(node.func, self.global_assigns)
|
|
if dotted == "fluent.migrate.helpers.transforms_from":
|
|
return self.call_helpers_transforms_from(node)
|
|
if dotted.startswith("fluent.migrate."):
|
|
return self.call_transform(node, dotted)
|
|
self.generic_visit(node)
|
|
|
|
def call_ctx(self, node):
|
|
if node.func.attr == "add_transforms":
|
|
return self.call_add_transforms(node)
|
|
raise BadContextAPIException(
|
|
"Unexpected call on {}.{}".format(self.ctx_var, node.func.attr)
|
|
)
|
|
|
|
def call_add_transforms(self, node):
|
|
args_msg = (
|
|
"Expected arguments to {}.add_transforms: "
|
|
"target_ftl_path, reference_ftl_path, list_of_transforms"
|
|
).format(self.ctx_var)
|
|
ref_msg = (
|
|
"Expected second argument to {}.add_transforms: "
|
|
"reference should be string or variable with string value"
|
|
).format(self.ctx_var)
|
|
# Just check call signature here, check actual types below
|
|
if not self.check_arguments(node, (ast.AST, ast.AST, ast.AST)):
|
|
self.issues.append(
|
|
{
|
|
"msg": args_msg,
|
|
"line": node.lineno,
|
|
}
|
|
)
|
|
return
|
|
in_reference = node.args[1]
|
|
if isinstance(in_reference, ast.Name):
|
|
in_reference = self.global_assigns.get(in_reference.id)
|
|
if isinstance(in_reference, ast.Constant):
|
|
in_reference = in_reference.value
|
|
if not isinstance(in_reference, str):
|
|
self.issues.append(
|
|
{
|
|
"msg": ref_msg,
|
|
"line": node.args[1].lineno,
|
|
}
|
|
)
|
|
return
|
|
self.references.add(in_reference)
|
|
# Checked node.args[1].
|
|
# There's not a lot we can say about our target path,
|
|
# ignoring that.
|
|
# For our transforms, we want more checks.
|
|
self.generic_visit(node.args[2])
|
|
|
|
def call_transform(self, node, dotted):
|
|
module, called = dotted.rsplit(".", 1)
|
|
if module not in ("fluent.migrate", "fluent.migrate.transforms"):
|
|
return
|
|
transform = getattr(transforms, called)
|
|
if not issubclass(transform, transforms.Source):
|
|
return
|
|
bad_args = f"{called} takes path and key as first two params"
|
|
if not self.check_arguments(
|
|
node,
|
|
(
|
|
(ast.Constant, ast.Name),
|
|
(ast.Constant, ast.Name),
|
|
),
|
|
allow_more=True,
|
|
check_kwargs=False,
|
|
):
|
|
self.issues.append({"msg": bad_args, "line": node.lineno})
|
|
return
|
|
path = node.args[0]
|
|
if isinstance(path, ast.Constant):
|
|
path = path.value
|
|
if isinstance(path, ast.Name):
|
|
path = self.global_assigns.get(path.id)
|
|
if not isinstance(path, PATH_TYPES):
|
|
self.issues.append({"msg": bad_args, "line": node.lineno})
|
|
|
|
def call_helpers_transforms_from(self, node):
|
|
args_msg = "Expected arguments to transforms_from: " "str, **substitions"
|
|
if not self.check_arguments(node, (ast.Constant,), check_kwargs=False):
|
|
self.issues.append(
|
|
{
|
|
"msg": args_msg,
|
|
"line": node.lineno,
|
|
}
|
|
)
|
|
return
|
|
kwargs = {}
|
|
found_bad_keywords = False
|
|
for keyword in node.keywords:
|
|
v = keyword.value
|
|
if isinstance(v, ast.Constant):
|
|
v = v.value
|
|
if isinstance(v, ast.Name):
|
|
v = self.global_assigns.get(v.id)
|
|
if isinstance(v, ast.Call):
|
|
v = "determined at runtime"
|
|
if not isinstance(v, PATH_TYPES):
|
|
msg = "Bad keyword arg {} to transforms_from".format(keyword.arg)
|
|
self.issues.append(
|
|
{
|
|
"msg": msg,
|
|
"line": node.lineno,
|
|
}
|
|
)
|
|
found_bad_keywords = True
|
|
else:
|
|
kwargs[keyword.arg] = v
|
|
if found_bad_keywords:
|
|
return
|
|
try:
|
|
transforms = transforms_from(node.args[0].value, **kwargs)
|
|
except MigrationError as e:
|
|
self.issues.append(
|
|
{
|
|
"msg": str(e),
|
|
"line": node.lineno,
|
|
}
|
|
)
|
|
return
|
|
ti = TransformsInspector()
|
|
ti.visit(transforms)
|
|
self.issues.extend(
|
|
{
|
|
"msg": issue,
|
|
"line": node.lineno,
|
|
}
|
|
for issue in set(ti.issues)
|
|
)
|
|
|
|
def check_arguments(self, node, argspec, check_kwargs=True, allow_more=False):
|
|
if check_kwargs and (
|
|
node.keywords or (hasattr(node, "kwargs") and node.kwargs)
|
|
):
|
|
return False
|
|
if hasattr(node, "starargs") and node.starargs:
|
|
return False
|
|
for arg, NODE_TYPE in zip_longest(node.args, argspec):
|
|
if NODE_TYPE is None:
|
|
return True if allow_more else False
|
|
if not (isinstance(arg, NODE_TYPE)):
|
|
return False
|
|
return True
|
|
|
|
|
|
class TransformsInspector(Visitor):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.issues = []
|
|
|
|
def generic_visit(self, node):
|
|
if isinstance(node, transforms.Source):
|
|
src = node.path
|
|
# Source needs paths to be normalized
|
|
# https://bugzilla.mozilla.org/show_bug.cgi?id=1568199
|
|
if src != mozpath.normpath(src):
|
|
self.issues.append(f'Source "{src}" needs to be a normalized path')
|
|
super().generic_visit(node)
|
|
|
|
|
|
def cli():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("migration")
|
|
args = parser.parse_args()
|
|
issues = Validator.validate(args.migration)["issues"]
|
|
for issue in issues:
|
|
# bright red fg
|
|
print(f"\x1b[1;31m{issue['msg']} at line {issue['line']}\x1b[0m")
|
|
return 1 if issues else 0
|