ref(examples): Add more examples to tests #182
Conversation
Also change how we clean up topics, so that a test can be run multiple times without cross-contamination. We should probably move to topic names with UUIDs in them though and stop messing with getsentry's topics
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| self.clock.sleep(1.0) | ||
| # Process any pending window closes | ||
| if self.processor is not None: | ||
| self.processor._run_once() |
There was a problem hiding this comment.
Bug: PipelineTestHarness Time Precision and Dependency Issues
The PipelineTestHarness class has two issues:
- The
advance_timemethod truncates thesecondsfloat parameter to an integer, losing fractional precision which can prevent windowing operations from triggering correctly. - It directly calls the private
_run_once()method ofStreamProcessor, creating a brittle dependency on Arroyo's internal implementation that may break.
| env: | ||
| FLINK_LIBS: ./flink_libs | ||
|
|
||
| integration-tests: |
There was a problem hiding this comment.
I'd like to keep this as a separate CI check, to avoid clutter.
There was a problem hiding this comment.
are you sure? to me this is just another kind of test and can be part of tests/
There was a problem hiding this comment.
I don't think this can replace @evanh 's integration test as it uses the Python Arroyo adapter. We do not support that adapter anymore. @evanh 's tests use the real prod runner with the rust adapter that we would use in prod.
While I like the applaud the idea of using the local broker more for tests I see two possible path forward:
- go on with this PR but reinstate @evanh 's test because the Python Arroyo adapter does not replace an integration test with the Rust Adapter we are actively developing.
- Support the local broker in the rust adapter and use that one.
| self.adapter = ArroyoAdapter.build( | ||
| { | ||
| "env": {}, | ||
| "steps_config": steps_config, | ||
| }, | ||
| consumers, | ||
| producers, | ||
| ) |
There was a problem hiding this comment.
This is not the runtime we would use in prod. It is also not maintained and is going to be deleted soon.
I don't think we can rely on this one for integration tests. @evanh 's previous implementation (by running the actual runner tested the code we would run in production, which is the rust adapter). We should not move back to the python one.
I do support the idea of a harness to test the pipeline end to end with the local broker and those can be considered standard tests, they do not have anything special that make them being a separate suite. Though:
- It has to run the code in the rust adapter or at least in one that we maintain.
- I still see a value in having 1-2 integration test where we start with the runner file and use a real Kafka. Those are a like Sentry acceptance tests.
There is a substantial difference between this test and @evanh's integration test: the adapter is not the same so the only relevant part is the pipeline code. Evan's test is actually testing end to end what SBC would use.
There was a problem hiding this comment.
okay, yeah you're right. would it be sufficient to move to rust-arroyo and its localbroker concept? not sure how exactly we'd do it but in principle my preferred solution to keep a real kafka broker out of this
i am not thrilled about having integration tests since they have been demonstrated to be really slow especially in this repo. i also think that we will at some point have to have a real story for testing in getsentry where integration tests are expensive.
There was a problem hiding this comment.
I would definitely prefer a local broker over the actual Kafka broker. I did it the way I did because I couldn't figure out a clean way to make rust arroyo + local broker work without significant refactoring.
There was a problem hiding this comment.
i am not thrilled about having integration tests since they have been demonstrated to be really slow especially in this repo.
I'd still think having 1 smoke test that uses a real kafka is valuable. It is a basic acceptance test, the last line of defense. One is enough.
For the rest I agree local broker is the way to go.
|
|
||
| extract_steps(pipeline) | ||
|
|
||
| # If extraction failed, use defaults |
There was a problem hiding this comment.
Why would this fail in a valid scenario ?
| # Setup ArroyoAdapter with LocalBroker | ||
| consumers: Dict[str, KafkaConsumer] = {} | ||
| producers: Dict[str, KafkaProducer] = {} | ||
| steps_config: Dict[str, Dict[str, Dict[str, Any]]] = {} |
There was a problem hiding this comment.
Have you considered subclassing the Adapter making it a TestingAdapter that uses localConsumer and LocalProducer instead of a real one. That should simplify this PR considerably.
| import importlib.util | ||
|
|
||
| # Dynamically import the pipeline module | ||
| spec = importlib.util.spec_from_file_location(f"example_{test.name}", test.pipeline_module) | ||
| if spec is None or spec.loader is None: | ||
| raise ImportError(f"Cannot load module from {test.pipeline_module}") | ||
|
|
||
| module = importlib.util.module_from_spec(spec) | ||
| spec.loader.exec_module(module) | ||
|
|
||
| # Get the pipeline from the module | ||
| pipeline = module.pipeline |
There was a problem hiding this comment.
I feel we could use this code in the runner as well and in other tests. Why not putting it in a module in the application code rather than test code ?
3 participants
we already use localbroker in one test, we can make the integration tests much faster and add more examples. some of the examples are failing but i don't have time to investigate and skipped them