Skip to content

Validation service

mongorunway.application.services.validation_service ¤

__all__: typing.Sequence[str] = ('validate_migration_process', 'validate_rule_dependencies_recursive') module-attribute ¤

validate_migration_process(migration_process, ctx) ¤

Source code in mongorunway\application\services\validation_service.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def validate_migration_process(
    migration_process: domain_migration.MigrationProcess,
    ctx: domain_context.MigrationContext,
) -> None:
    if migration_process.has_rules():
        _LOGGER.info(
            "Starting validation of migration process with version %s...",
            migration_process.migration_version,
        )

        for rule in migration_process.rules:
            if not rule.is_independent():
                validate_rule_dependencies_recursive(
                    ctx=ctx,
                    depends_on=rule.depends_on,
                )

            if rule.check_is_broken(ctx):
                _LOGGER.error("%s rule failed, raising...", rule.name)
                raise domain_exception.MigrationBusinessRuleBrokenError(rule)

            _LOGGER.info("%s rule successfully passed.", rule.name)

validate_rule_dependencies_recursive(depends_on, ctx) ¤

Source code in mongorunway\application\services\validation_service.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def validate_rule_dependencies_recursive(
    depends_on: typing.Sequence[domain_rule.MigrationBusinessRule],
    ctx: domain_context.MigrationContext,
) -> None:
    for rule in depends_on:
        if rule.check_is_broken(ctx):
            _LOGGER.error("%s rule failed, raising...", rule.name)
            raise domain_exception.MigrationBusinessRuleBrokenError(rule)

        _LOGGER.info("%s rule successfully passed.", rule.name)

        if rule.is_independent():
            continue

        validate_rule_dependencies_recursive(
            ctx=ctx,
            depends_on=rule.depends_on,
        )