Inject failures to find resource leaks

Goal

You observe normal, repeated-shutdown, and failure paths, and the startup and cleanup of the app lifespan.

Why it matters

There was code that closed the connection when the request succeeded, but when an exception occurred the open connection was left behind. A test that checks only success responses did not reveal this leak. A lifecycle test has to focus on when the resource was opened and closed rather than on the return value.

Steps

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: new_resource() is a new dictionary {open:False, events:[]}, and separate calls do not share events. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Prepare this once at the start. Existing files are not overwritten.

mkdir -p /root/work/test-lifespan-cleanup-lab
test -e /root/work/test-lifespan-cleanup-lab/service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/service.py /root/work/test-lifespan-cleanup-lab/service.py
test -e /root/work/test-lifespan-cleanup-lab/test_service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/test_service.py /root/work/test-lifespan-cleanup-lab/test_service.py
cd /root/work/test-lifespan-cleanup-lab

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: start(resource) raises ValueError if it is already open; otherwise it sets open=True and appends 'open' to events. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.
In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: stop(resource) sets open=False and appends 'close' to events only when it is open. If it is already closed, it leaves things as they are. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.
In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: read(resource) raises RuntimeError if it is closed, and returns {ready:True} if it is open. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.
In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: scope(resource) is a contextmanager. On entry it calls start, inside the block it yields resource, and on both success and failure of the block it closes with stop. An exception in the block is propagated. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.
In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: lifespan_for(resource) returns an asynccontextmanager function lifespan(app). Inside scope(resource), it sets app.state.resource and yields. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.
In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: create_app(resource) uses lifespan_for. GET /ready returns the result of read on app.state.resource. The resource must be closed when the context ends. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.
In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: exercise(resource, fail=False) calls GET /ready inside with TestClient(create_app(resource)). If fail=True, it raises RuntimeError inside, and otherwise it returns the response JSON. In both cases the resource must be closed. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Notes

You work in the existing lab-dev environment with no internet and no package installation.
Each step runs within a 45-second grading budget. Do not add real sleeps or network calls.
The submitted tests are run in a separate temporary folder against the correct and defective implementations. Against the correct one, every test that actually runs must pass, and against a defective one, the body of a test must fail. A collection error, zero tests run, everything skipped, and forced termination are not a pass. Use only the basic pytest features and the provided libraries.
FastAPI official documentation · pytest official documentation · Python sqlite3
Limitation: The teaching resource dictionary is an observation device that stands in for a real DB connection pool. In production you also have to design for partial initialization failure, concurrency of the connection pool, cancellation handling, and a shutdown time limit. Instead of writing event strings in a report, you check the state of objects that the learner's code changed as it ran. You may read the provided implementation, but grading uses a separate copy. Do not work around a defect by checking the wording of the source or by modifying files; check the execution results of the public interface.

Make the resource state independent — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: new_resource() is a new dictionary {open:False, events:[]}, and separate calls do not share events. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Prepare this once at the start. Existing files are not overwritten.

mkdir -p /root/work/test-lifespan-cleanup-lab
test -e /root/work/test-lifespan-cleanup-lab/service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/service.py /root/work/test-lifespan-cleanup-lab/service.py
test -e /root/work/test-lifespan-cleanup-lab/test_service.py || cp /opt/fixtures/ten_labs/test-lifespan-cleanup-lab/test_service.py /root/work/test-lifespan-cleanup-lab/test_service.py
cd /root/work/test-lifespan-cleanup-lab

Do not share a mutable list through a global or a default argument. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/01-contract.sh.

Reject a duplicate start — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: start(resource) raises ValueError if it is already open; otherwise it sets open=True and appends 'open' to events. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Reject the behavior of starting twice and losing one resource. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/02-contract.sh.

Make shutdown idempotent — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: stop(resource) sets open=False and appends 'close' to events only when it is open. If it is already closed, it leaves things as they are. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

Even when several cleanup paths overlap, a duplicate close event must not appear. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/03-contract.sh.

Block the use of a closed resource — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: read(resource) raises RuntimeError if it is closed, and returns {ready:True} if it is open. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

The ready state and the existence of the object are different things. The object can exist and still be closed. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/04-contract.sh.

Put finally on the exception path — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: scope(resource) is a contextmanager. On entry it calls start, inside the block it yields resource, and on both success and failure of the block it closes with stop. An exception in the block is propagated. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

If you write the close only after the yield, that line is never reached when an exception occurs. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/05-contract.sh.

Connect the app lifespan to the resource — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: lifespan_for(resource) returns an asynccontextmanager function lifespan(app). Inside scope(resource), it sets app.state.resource and yields. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

You do not call the lifespan function itself; you pass it to the FastAPI constructor. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/06-contract.sh.

Read the ready state through a real request — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: create_app(resource) uses lifespan_for. GET /ready returns the result of read on app.state.resource. The resource must be closed when the context ends. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

You have to use with TestClient for both the lifespan startup and shutdown to run. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/07-contract.sh.

Clean up after a failure following the request — test

In /root/work/test-lifespan-cleanup-lab/test_service.py, test the following public contract of the provided service.py: exercise(resource, fail=False) calls GET /ready inside with TestClient(create_app(resource)). If fail=True, it raises RuntimeError inside, and otherwise it returns the response JSON. In both cases the resource must be closed. It must pass against the correct implementation and be caught, through a failure in the body of an actual test, in an implementation that breaks this contract. Keep the tests from the earlier steps and add a test_ function.

If you bind the normal path and the exception path into the same cleanup structure, you reduce the number of shutdown paths you miss. Do not modify the implementation file. Use pytest.raises to check the expected exception, and assert a concrete expected value for the normal result.

After saving, check with bash /opt/lab/checks/test-lifespan-cleanup-lab/08-contract.sh.