authentik.flows.tests.test_executor.TestFlowExecutor::test_reevaluate_remove_consecutive

Stack Traces | 0.166s run time

self = <unittest.case._Outcome object at 0x7f99bdced0f0>
test_case = <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>
result = <TestCaseFunction test_reevaluate_remove_consecutive>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
                    self._callSetUp()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>
method = <bound method TestFlowExecutor.test_reevaluate_remove_consecutive of <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>>

    def _callTestMethod(self, method):
>       if method() is not None:

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>

    def test_reevaluate_remove_consecutive(self):
        """Test planner with re-evaluate (consecutive stages are removed)"""
        flow = create_test_flow(
            FlowDesignation.AUTHENTICATION,
        )
        false_policy = DummyPolicy.objects.create(
            name=generate_id(), result=False, wait_min=1, wait_max=2
        )
    
        binding = FlowStageBinding.objects.create(
            target=flow,
            stage=DummyStage.objects.create(name=generate_id()),
            order=0,
            evaluate_on_plan=True,
            re_evaluate_policies=False,
        )
        binding2 = FlowStageBinding.objects.create(
            target=flow,
            stage=DummyStage.objects.create(name=generate_id()),
            order=1,
            re_evaluate_policies=True,
        )
        binding3 = FlowStageBinding.objects.create(
            target=flow,
            stage=DummyStage.objects.create(name=generate_id()),
            order=2,
            re_evaluate_policies=True,
        )
        binding4 = FlowStageBinding.objects.create(
            target=flow,
            stage=DummyStage.objects.create(name=generate_id()),
            order=2,
            evaluate_on_plan=True,
            re_evaluate_policies=False,
        )
    
        PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
        PolicyBinding.objects.create(policy=false_policy, target=binding3, order=0)
    
        # Here we patch the dummy policy to evaluate to true so the stage is included
        with patch("authentik.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE):
            exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
            # First request, run the planner
            response = self.client.get(exec_url)
            self.assertEqual(response.status_code, 200)
            self.assertStageResponse(response, flow, component="ak-stage-dummy")
    
            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
    
            self.assertEqual(plan.bindings[0], binding)
            self.assertEqual(plan.bindings[1], binding2)
>           self.assertEqual(plan.bindings[2], binding3)

.../flows/tests/test_executor.py:516: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>
first = <FlowStageBinding: Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502>
second = <FlowStageBinding: Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502>
msg = None

    def assertEqual(self, first, second, msg=None):
        """Fail if the two objects are unequal as determined by the '=='
           operator.
        """
        assertion_func = self._getAssertEqualityFunc(first, second)
>       assertion_func(first, second, msg=msg)

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:907: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.flows.tests.test_executor.TestFlowExecutor testMethod=test_reevaluate_remove_consecutive>
first = <FlowStageBinding: Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502>
second = <FlowStageBinding: Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502>
msg = '<Flow[13 chars] Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502> != <Flow[13 chars] Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502>'

    def _baseAssertEqual(self, first, second, msg=None):
        """The default assertEqual implementation, not type specific."""
        if not first == second:
            standardMsg = '%s != %s' % _common_shorten_repr(first, second)
            msg = self._formatMessage(msg, standardMsg)
>           raise self.failureException(msg)
E           AssertionError: <Flow[13 chars] Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502> != <Flow[13 chars] Flow-stage binding #2 to 0d76a76b-03a7-4038-b3c5-428570a79502>

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:900: AssertionError