Ensure Transforms and Checks are executed with the correct timeout

[dgarros]

Fixes #5456

This PR enforces the timeout for Transforms and Checks leveraging Prefect timeout capabilities.
One good learning from this exercice is that we can setup custom timeout when we are calling a flow or a task directly but it's not possible when using a deployment.

I was able to manually validate that it's working as expected but I couldn't come up with a good way to build a unit tests for timeout.

I also fixed a few minor issues related to flow_run_name and tags and in a few places I introduced Protocols to improve typing

[codspeed-hq]

CodSpeed Performance Report

Merging #5471 will not alter performance

_{Comparing dga-20240115-fix-5456 (d3561ec) with stable (5e914b2)}

Summary

✅ 10 untouched benchmarks

[LucasG0]

I was able to manually validate that it's working as expected but I couldn't come up with a good way to build a unit tests for timeout.

I think one way could be to patch flow.fn attribute of a given flow with an asyncio.sleep(timeout + 1), then call flow.with_options(timeout=timeout)() and expect it to raise a TimeoutError

[ogenstad]

rfiles, oh the memories :)

[ogenstad]

Based on #5483 and opsmill/infrahub-sdk-python#55 I think we should look at moving the timeout value away from this file and instead define it in .infrahub.yml instead (so that it's also possible to do this with a Jinja2 Transform). I'm wondering when we make that change what the impact would be with this setting here. Do we need a temporary test that validates that the timeout of this Transform is set to 2 just so that it doesn't just slip through the cracks if we update it later. I.e. I think we'll forget to go back here and remove it and update the .infrahub.yml file for this repo.

[dgarros]

This is a good point but not sure if we can easily add this test today, I don't think this repo fixtures is used to test the import of Python Transform, at least I couldn't find it.

[ogenstad]

It's imported here:

class TestProposedChangePipelineRepository(TestInfrahubApp):
    @pytest.fixture(scope="class")
    async def initial_dataset(
        self,
        db: InfrahubDatabase,
        initialize_registry: None,
        git_repos_source_dir_module_scope: Path,
        client: InfrahubClient,
        bus_simulator: BusSimulator,
        prefect_test_fixture,
    ) -> None:
        await load_schema(db, schema=CAR_SCHEMA)

        bus_simulator.service.cache = RedisCache()

        john = await Node.init(schema=TestKind.PERSON, db=db)
        await john.new(db=db, name="John", height=175, age=25, description="The famous Joe Doe")
        await john.save(db=db)
        koenigsegg = await Node.init(schema=TestKind.MANUFACTURER, db=db)
        await koenigsegg.new(db=db, name="Koenigsegg")
        await koenigsegg.save(db=db)
        people = await Node.init(schema=InfrahubKind.STANDARDGROUP, db=db)
        await people.new(db=db, name="people", members=[john])
        await people.save(db=db)

        jesko = await Node.init(schema=TestKind.CAR, db=db)
        await jesko.new(
            db=db,
            name="Jesko",
            color="Red",
            description="A limited production mid-engine sports car",
            owner=john,
            manufacturer=koenigsegg,
        )
        await jesko.save(db=db)

        branch1 = await client.branch.create(branch_name="branch1")

        FileRepo(name="car-dealership", sources_directory=git_repos_source_dir_module_scope)
        client_repository = await client.create(
            kind=InfrahubKind.REPOSITORY,
            data={"name": "car-dealership", "location": f"{git_repos_source_dir_module_scope}/car-dealership"},
            branch=branch1.name,
        ) <---
        await client_repository.save()

I'm not sure what happens if we don't change this if something starts failing or if we'd just add 8 seconds to the test time and not realize why?