Closed rohansh-tty closed 1 year ago
So the test I promised. Put this to the rocketry\test\app\test_app.py
.
def test_delete_task():
set_logging_defaults()
app = Rocketry(config={'task_execution': 'main'})
@app.task()
def task_1():
...
@app.task(name="task 2")
def task_2():
...
assert len(app.session.tasks) == 2
app.session.remove_task(task_1)
assert len(app.session.tasks) == 1
app.session.remove_task("task 2")
assert len(app.session.tasks) == 0
And actually, I realized this fix is not enough though as if someone passes the task function it will still fail.
But this should fix it:
def remove_task(self, task: Union['Task', str]):
from rocketry.core.task import Task
if not isinstance(task, Task):
task = self[task]
self.tasks.remove(task)
There is a risk of circular imports so not sure if Task
can be imported where the other imports are.
I added the test case and modified the remove_task() method. Also, I didn't face any issues related to circular imports. Here's a quick snap of test case results.
Ye, looks good! Moreover, those warnings are most likely just that you are missing some pytest plugins. The CI will fail if warnings are raised there.
I'll try to merge this today or latest tomorrow. Thanks for seeing the effort to fix this!
Ye, Thanks @Miksus for helping me out. Also, I'll install the missing pytest plugins.
Minor change in session module's remove_task() method.