Beyond Hardcoding: My Breakthrough in Testable Parallel Python

Last week, we talked about removing hardcoded synchronization primitives. The refactoring was prompted by my revision to a take-home assignment I submitted for a job application. This week, let’s get into testing, while building a smaller-scale replica, building on what we learned. Read on to join me in a self-reflecting journey regarding take-home assignment and why I believe they often fail to serve their intended purpose.

A cute illustration by Copilot on the topic

Beyond the Code: The True Cost of Interview Assignments

One may ask, what’s the deal with all these take-home assignments that software engineers are required to take when we apply for a job? Why do engineers seem to not like them, but still debate endlessly on the topic? How does it work?

I considered myself lucky to only start encountering them in my current round of job applications. Though not uncommon in the past; they weren’t yet a widespread practice. For a backend developer like myself, the assignments typically require implementing a set of CRUD endpoints, sometimes involving some state management. Under normal working conditions, the whole process usually takes about 2–3 days. The time spent on writing code is usually rather short, compared to the time spent on setup and testing. The orchestration of various components like the database, cache, backend application, and server to ensure seamless integration is quite time-consuming.

If one is applying for a full-stack position, add frontend compilation and server setup to the list of tasks too.

Photo by Jefferson Santos on Unsplash

Most of the time, the hiring companies don’t offer any compensation upon submission. Spending multiple days on an assignment, then, is actually economically impractical. Regardless of how trivial the requirement reads, it still mimics real-world work to a certain extent. On one occasion, a company even gave me a requirement list that covered my future work with them, if I were hired. Hiring companies sometimes specify protocols and tech stacks, requiring extra learning time.

Some may wonder, won’t LLM be useful in this?

To a lot of engineers, LLM-empowered editors are already an integral part of their daily work. Though they could be useful, these editors usually require a subscription. Given the reality where the hiring company is already not paying for the work, subscribing to these services just for these assignments makes no economic sense. On the other hand, LLMs are not that useful in the setup and integration work.

Vibe-Coding, Storytelling, and LLMs: A Collaborative Approach

The Perils of Unstated Requirements

Regardless, take-home assignment can still be a good way to gauge a candidate’s ability. How are they executed so poorly that engineers detest them so passionately over social media? Seriously, I once posted a random rant over an extensive assignment requirement at LinkedIn, and that post made hundreds of thousands impressions and sparked a heated discussion on whether I should proceed.

Let’s look into the assignment we are discussing today, it was done for a job application referred by a contact. The scope of the project seemed manageable, and looked doable within a day. Background processing was not listed as a requirement, but I opted to include that in my submission anyway. Though the offer was unsatisfactory, I attempted anyway mainly to demonstrate my respect and appreciation to the referrer.

Little did I know, my oversight regarding the hardcoded queue caused much complication for the tests.

Life went on as usual after the submission, until the company replied with the same undesirable offer. Interestingly, the examiner noted that JWT was not implemented, an indication of my inability to adhere to requirements. Conversely, the extra effort on moving processing to background, auto container image building was barely acknowledged.

Photo by Wes Hicks on Unsplash

Naturally, that feedback triggered dissatisfaction and prompted me to revisit the original requirement document.

And no, there was no mention of JWT.

On my LLM-assistant’s advice, I politely declined the offer and did not attempt to elaborate on the omission of JWT in the initial requirement. Still, this highlights a problem — bad communication between the examiner and the applicant. Unlike in a daily work setting, where one has better access to coworkers or clients for clarifications. It is hard enough to get prompt responses from the hiring company these days, and this makes the take-home assignment work more like a one-off engagement. Requirement is sent, followed by a submission, with little room for revision and corrections.

If life is a role playing game (RPG), I probably need to consider dumping more points into my Intelligence to improve my psychic ability.

On the other hand, why can’t the hiring company communicate better?

Why penalize the candidate for not meeting implicit criteria? Funnily enough (OK, I was absolutely frustrated at the time), I was once rejected for trivial, random refactoring decisions despite the project worked as specified. The feedback was almost laughable as if they were joking, as it covered flaws such as certain classes should be stored in another module, incorrect pluralization handling of classes and table names, the omission of explicit module imports, and they even gave me a fail when they were unable to locate my frontend implementation. The feedback almost gave me an impression that the company doesn’t do peer review, suggesting I would be left to work on projects independently.

And yes, the contact person disappeared even after I pointed the actual line where the supposedly missing line is.

From Hardcoding Headaches to Testable Triumph

Photo by AbsolutVision on Unsplash

Why revisit the project then? Despite some minor problems with the tests, the project worked. Why undertake such thorough refactoring to remove the hardcoding, for a one-off assignment?

To a certain extent, it is the revelation that matters. The refactoring (which culminated in this revelation) was discussed last week, and this week we are going to see how passing synchronization primitives (like a queue) explicitly in a parallel programming setting improves testability.

Practically speaking, despite the experimental nature, BigMeow is my first formal attempt at asynchronous programming in a parallel setting. After spending so much time and effort in learning and re-learning AsyncIO with it, it is now becoming my go-to reference and template for projects requiring asynchronous and parallel setup. Moreover, I wrote a series of posts on asynchronous programming (here, here and here) based on the code written for BigMeow.

The revelation came when I was adding a scheduler to the chatbot: What if I remove the hardcoding on the synchronization primitives? Fixing it in BigMeow right away would have been a huge undertaking, so I went to the assignment with simpler scope. Structurally, both are similar, as the assignment practically shared much of its core structure with BigMeow. That worked right away, and as a bonus, it fixed all the unexplained problems I had with tests.

Oh, the hours I wasted in stress trying to fix the tests.

Refactoring on BigMeow was equally successful, though a lot more involved given the complexity. It is interesting to see how a sudden flash of insight impacts the thought process and changes how code is written.

Hands-On: Crafting Decoupled Background Processes

Today we are revisiting the assignment together, which is yet another unimaginative FastAPI prototype accompanied by a test suite. The technique detailed in the previous article is used to remove the hardcoded synchronization primitives. That change makes testing much more intuitive. In order to keep this concise, we are not implementing the implicit JWT requirement.

As shown in the article last week, we start from the main process, beginning with the setup of a ProcessPoolExecutor.

from project import web, background
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp

# create the synchronization primitives
manager = mp.Manager()
exit_event = manager.Event()
task_queue = manager.Queue()

# start the processes in parallel
with ProcessPoolExecutor() as executor:
    executor.submit(web.run, exit_event, task_queue)
    executor.submit(background.run, exit_event, task_queue)

All the synchronization primitives are sent individually to the run() function this time, given the simplified scope of this article. Additionally we are omitting graceful shutdown for brevity, feel free to refer to our previous discussion on the topic.

Similarly, the setup for the FastAPI web module would look like this

@asynccontextmanager
async def lifespan(app: FastAPI):
    if not hasattr(app.state, "task_queue") and isinstance(
        app.state.task_queue, Queue
    ):
        raise RuntimeError("Task queue is missing")

        yield

app = FastAPI(lifespan=lifespan)
async def run(exit_event: Event, task_queue: Queue) -> None:
    app.state.task_queue = task_queue
    server = uvicorn.Server(...)
    asyncio.create_task(server.serve())
    await asyncio.to_thread(exit_event.wait)
    await server.shutdown()

Nothing new here, it was almost the exact code we just saw last week. Next comes the crucial part, where we are actually receiving a batch of data:

@dataclass
class SomeData:
    id: str
    num: int

@app.post("/submit/batch")
async def claim_submit(
    request: Request,
    : list[SomeData],
) -> None:
    await asyncio.to_thread(
        request.app.state.batch_queue.put, submitted
    )

Just assume we are returning a 200 OK every time without a body. After that, we can start by setting up a fixture for our TestClient

import pytest
from fastapi.testclient import TestClient

from project.web import app

@pytest.fixture(name="client")
def client_fixture():
    app.state.task_queue = multiprocessing.Manager().Queue()
    with TestClient(app) as client:
        yield client

With the task_queue set, we can write a test to see if the queue is receiving submitted data

@pytest.mark.asyncio
async def test_create_job(client: TestClient):
    submitted = [
        {"id": "RECORD01", "num": 1},
        {"id": "RECORD01", "num": 3},
        {"id": "RECORD01", "num": 5},
        {"id": "RECORD01", "num": 7},
        {"id": "RECORD01", "num": 15},
    ]

    response = client.post("/submit/batch", json=submitted)

    assert response.status_code == 200

    # check if the job is sent to the queue
    payload = client.app.state.batch_queue.get()

    for idx, record in enumerate(submitted):
        assert record['id'] == payload[idx].id
        assert record['num'] == payload[idx].num

The background processing module I submitted was very basic, it was mostly just a proof-of-concept based on the provided spec. For this simplified example, let’s just do a Fizz Buzz check on the submitted num. Similar to the web module, let’s start with the setup:

def run(exit_event: Event, task_queue: Queue):
    database = create_a_fictional_database_connection()

    asyncio.create_task(loop_queue(task_consume, task_queue, database))

    await asyncio.to_thread(exit_event.wait)

Firstly, we have the actual function processing the batch data, and the result is saved through a fictional database.save method,

async def process_payload(database, payload: list[ChargeIncoming]) -> None:
    result = []

    for record in payload:
        fizz = "fizz" if record.num % 3 == 0 else ""
        buzz = "buzz" if record.num % 5 == 0 else ""
        result.append(f"{fizz}{buzz}" or str(record.num))
    await database.save(result)

And the corresponding test would look something like, assume we have a fictional database fixture defined somewhere:

@pytest.mark.asyncio
async def test_process_payload(database):
    payload = [
        SomeData("RECORD01", 1),
        SomeData("RECORD01", 3),
        SomeData("RECORD01", 5),
        SomeData("RECORD01", 7),
        SomeData("RECORD01", 15),
    ]

    expected = [
        "1",
        "fizz",
        "buzz",
        "7",
        "fizzbuzz"
    ]

    await process_payload(database, payload)

    # test if the processing logic is done properly
    assert database.exists(expected)

Lastly, the queue consumption would look like

async def loop_queue(func: Callable[..., Awaitable[None]], *args: Any):
    while True:
        func(*args)

async def task_consume(task_queue: Queue, database):
    payload = await asyncio.create_task(
        asyncio.to_thread(task_queue.get, timeout=5.0)
    )

    await process_payload(database, payload)

Now the corresponding test

@pytest.mark.asyncio
async def test_task_consume(database):
    payload = [
        SomeData("RECORD01", 1),
        SomeData("RECORD01", 3),
        SomeData("RECORD01", 5),
        SomeData("RECORD01", 7),
        SomeData("RECORD01", 15),
    ]

    queue = multiprocessing.Manager().Queue()

    queue.put(payload)

    with patch('background.process_payload') as process_payload:
        await task_consume(task_queue, database)
        process_payload.assert_called_once_with(database, payload)

Comparing the tests for the FastAPI application and the background module, the benefit is more apparent in the latter case. With the explicit passing of the queue object as an argument, there’s no more guesswork on which queue to patch in test. Plus, as we are ensuring both the test and application are referring to the same queue now, we can pass real data into test, and ensure they are being received properly.

Impact & Insights: From Personal Challenge to Shared Knowledge

So my submitted solution showcased a way to process data in batch, in a separate module running in parallel. Mainly due to the time limit, and the scale of the exercise, I chose to propose a very basic implementation. The replacement of the queue can be done fairly easily with a more robust solution. In my humble opinion, the point of the exercise should be on the comprehension of requirement, and deliver a solution within a timeframe both parties find reasonable.

Photo by Akson on Unsplash

Documenting the insight into blog posts can be seen as my attempt to think out loud. Who knows if this could spark further discussion that yields a better design for similar setup? Granted, the length of this article exceeded my initial expectation, but understanding the context is also crucial to know how it leads to the revelation. Perhaps some people may appreciate the thought process too?

Earlier, I talked about adding a scheduler to my chatbot BigMeow. It took me quite a while to get it working in this asynchronous parallel setup. Perhaps we can go through the process of adapting it to our setup in a more generalized form in a future article. Stay tuned if you are interested in the topic.

At last, thank you again for reading this far, and I shall write again, next week.

For this article, I received invaluable editorial assistance from Gemini, my AI assistant. While Gemini helped refine the language and structure, please know that the ideas, personal voice, and all code snippets are entirely my own. If you’re interested in collaborating on a project or exploring job opportunities, feel free to connect with me here on Medium or reach out on LinkedIn.