Repository: mamba-org/quetz (language: python)

Issue:
show tasks triggered in actions under /jobs endpoints
Trigger task with:

```
curl -X PUT localhost:8000/api/channels/conda-forge/actions -d '{"action":"generate_indexes"}' -H "X-Api-key: ${QUETZ_API_KEY}"
```

monitor execution with:

```
curl localhost:8000/api/jobs -H "X-api-key: ${QUETZ_API_KEY}"
```

and 

```
curl localhost:8000/api/jobs/{job_id}/tasks -H "X-api-key: ${QUETZ_API_KEY}"
```

TODO:

- [x] add action task to jobs/tasks
- [x] change job/task status after action failed or succeeded
- [x] migrate all actions
- [x] make sure that watch-job-queue works
- [x] set task status from threading workers
- [x] add tests
- [x] return job_id from actions endpoint
- [x] check channel owner permissions to check the job/task status



Discussion / hints:
# [Codecov](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=h1) Report
> Merging [#364](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=desc) (ebf790f) into [master](https://codecov.io/gh/mamba-org/quetz/commit/fd6a70bcecaf100f3112858f8dfccf4429ca9e89?el=desc) (fd6a70b) will **decrease** coverage by `0.73%`.
> The diff coverage is `86.09%`.

[![Impacted file tree graph](https://codecov.io/gh/mamba-org/quetz/pull/364/graphs/tree.svg?width=650&height=150&src=pr&token=Z9RZXN9BBF)](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #364      +/-   ##
==========================================
- Coverage   84.73%   83.99%   -0.74%     
==========================================
  Files          68       70       +2     
  Lines        5182     5243      +61     
==========================================
+ Hits         4391     4404      +13     
- Misses        791      839      +48     
```


| [Impacted Files](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [quetz/authorization.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovYXV0aG9yaXphdGlvbi5weQ==) | `94.32% <33.33%> (-0.68%)` | :arrow_down: |
| [quetz/tasks/common.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovdGFza3MvY29tbW9uLnB5) | `74.60% <58.33%> (-0.87%)` | :arrow_down: |
| [quetz/jobs/models.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovam9icy9tb2RlbHMucHk=) | `98.03% <80.00%> (-1.97%)` | :arrow_down: |
| [quetz/jobs/runner.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovam9icy9ydW5uZXIucHk=) | `92.25% <87.50%> (-0.94%)` | :arrow_down: |
| [quetz/tasks/workers.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovdGFza3Mvd29ya2Vycy5weQ==) | `84.65% <87.50%> (+0.90%)` | :arrow_up: |
| [quetz/jobs/rest\_models.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovam9icy9yZXN0X21vZGVscy5weQ==) | `88.88% <88.88%> (ø)` | |
| [quetz/jobs/api.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovam9icy9hcGkucHk=) | `94.11% <90.90%> (+1.17%)` | :arrow_up: |
| [quetz/dao.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovZGFvLnB5) | `90.15% <100.00%> (+0.04%)` | :arrow_up: |
| [quetz/deps.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovZGVwcy5weQ==) | `82.97% <100.00%> (ø)` | |
| [quetz/jobs/dao.py](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree#diff-cXVldHovam9icy9kYW8ucHk=) | `100.00% <100.00%> (ø)` | |
| ... and [5 more](https://codecov.io/gh/mamba-org/quetz/pull/364/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=footer). Last update [fd6a70b...ebf790f](https://codecov.io/gh/mamba-org/quetz/pull/364?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Fix patch:
diff --git a/quetz/jobs/runner.py b/quetz/jobs/runner.py
index 8d822e3e..9948176a 100644
--- a/quetz/jobs/runner.py
+++ b/quetz/jobs/runner.py
@@ -144,6 +144,8 @@ def run_jobs(db, job_id=None, force=False):
             # so skipping here
             continue
 
+        job.status = JobStatus.running
+
         task = None
         for version in q:
             task = Task(job=job, package_version=version)
@@ -152,8 +154,9 @@ def run_jobs(db, job_id=None, force=False):
             logger.warning(
                 f"no versions matching the package spec {job.items_spec}. skipping."
             )
-            # job.status = JobStatus.success
-        job.status = JobStatus.running
+            if job.items_spec:
+                # actions have no related package versions
+                job.status = JobStatus.success
     db.commit()
 
 
@@ -162,11 +165,12 @@ def add_task_to_queue(db, manager, task, *args, func=None, **kwargs):
 
     if func is None:
         func = task.job.manifest
-
-    job = manager.execute(func, *args, **kwargs)
-    _process_cache[task.id] = job
     task.status = TaskStatus.pending
     task.job.status = JobStatus.running
+    db.add(task)
+    db.commit()
+    job = manager.execute(func, *args, task_id=task.id, **kwargs)
+    _process_cache[task.id] = job
     return job
 
 
@@ -204,44 +208,13 @@ def run_tasks(db, manager):
 
 
 def check_status(db):
-    tasks = db.query(Task).filter(
-        Task.status.in_([TaskStatus.running, TaskStatus.pending])
+    tasks = (
+        db.query(Task)
+        .filter(Task.status.in_([TaskStatus.running, TaskStatus.pending]))
+        .filter(Task.package_version_id.isnot(None))
     )
-    try:
-        for task in tasks:
-            if not task.package_version:
-                # task was launched from actions
-                continue
-            try:
-                process = _process_cache[task.id]
-            except KeyError:
-                logger.warning(f"running process for task {task} is lost, restarting")
-                task.status = TaskStatus.created
-                continue
-            status = process.status
-            if status == "pending":
-                task.status = TaskStatus.pending
-            elif status == "running":
-                task.status = TaskStatus.running
-            elif status == "failed":
-                task.status = TaskStatus.failed
-                _process_cache.pop(task.id)
-            elif status == "success":
-                task.status = TaskStatus.success
-                _process_cache.pop(task.id)
-
-        # update jobs with all tasks finished
-        (
-            db.query(Job)
-            .filter(Job.status.in_([JobStatus.running, JobStatus.pending]))
-            .filter(
-                ~Job.tasks.any(
-                    Task.status.in_(
-                        [TaskStatus.running, TaskStatus.pending, TaskStatus.created]
-                    )
-                )
-            )
-            .update({"status": JobStatus.success}, synchronize_session=False)
-        )
-    finally:
-        db.commit()
+    for task in tasks:
+        if task.id not in _process_cache:
+            logger.warning(f"running process for task {task} is lost, restarting")
+            task.status = TaskStatus.created
+    db.commit()


Test patch:
diff --git a/quetz/tests/api/conftest.py b/quetz/tests/api/conftest.py
index a65cc57b..c2aec5e5 100644
--- a/quetz/tests/api/conftest.py
+++ b/quetz/tests/api/conftest.py
@@ -84,6 +84,8 @@ def make_package_version(
 
     pkgstore = config.get_package_store()
 
+    versions = []
+
     def _make_package_version(filename, version_number, platform="linux-64"):
         filename = Path(filename)
         with open(filename, "rb") as fid:
@@ -106,19 +108,22 @@ def _make_package_version(filename, version_number, platform="linux-64"):
 
         dao.update_channel_size(channel_name)
 
+        versions.append(version)
+
         return version
 
     yield _make_package_version
 
+    for version in versions:
+        db.delete(version)
+    db.commit()
+
 
 @pytest.fixture
 def package_version(db, make_package_version):
     version = make_package_version("test-package-0.1-0.tar.bz2", "0.1")
 
-    yield version
-
-    db.delete(version)
-    db.commit()
+    return version
 
 
 @pytest.fixture()
diff --git a/quetz/tests/api/test_channels.py b/quetz/tests/api/test_channels.py
index a56bc091..895a2f9f 100644
--- a/quetz/tests/api/test_channels.py
+++ b/quetz/tests/api/test_channels.py
@@ -185,7 +185,15 @@ def test_get_channel_members(auth_client, public_channel, expected_code):
     assert response.status_code == expected_code
 
 
-def test_channel_names_are_case_insensitive(auth_client, maintainer):
+@pytest.fixture
+def remove_package_versions(db):
+    yield
+    db.query(db_models.PackageVersion).delete()
+
+
+def test_channel_names_are_case_insensitive(
+    auth_client, maintainer, remove_package_versions
+):
 
     channel_name = "MyChanneL"
 
@@ -333,7 +341,12 @@ def test_accents_make_unique_channel_names(auth_client, maintainer):
 
 
 def test_upload_package_version_to_channel(
-    auth_client, public_channel, maintainer, db, config
+    auth_client,
+    public_channel,
+    maintainer,
+    db,
+    config,
+    remove_package_versions,
 ):
     pkgstore = config.get_package_store()
 
diff --git a/quetz/tests/api/test_main_packages.py b/quetz/tests/api/test_main_packages.py
index 1258a7c4..0e853b59 100644
--- a/quetz/tests/api/test_main_packages.py
+++ b/quetz/tests/api/test_main_packages.py
@@ -227,9 +227,21 @@ def test_get_non_existing_package_version(
     assert response.status_code == 404
 
 
+@pytest.fixture
+def remove_package_versions(db):
+    yield
+    db.query(PackageVersion).delete()
+
+
 @pytest.mark.parametrize("package_name", ["test-package", "my-package"])
 def test_upload_package_version(
-    auth_client, public_channel, public_package, package_name, db, config
+    auth_client,
+    public_channel,
+    public_package,
+    package_name,
+    db,
+    config,
+    remove_package_versions,
 ):
     pkgstore = config.get_package_store()
 
@@ -332,7 +344,7 @@ def test_package_name_length_limit(auth_client, public_channel, db):
     assert pkg is not None
 
 
-def test_validate_package_names(auth_client, public_channel):
+def test_validate_package_names(auth_client, public_channel, remove_package_versions):
 
     valid_package_names = [
         "interesting-package",
@@ -374,7 +386,13 @@ def test_validate_package_names(auth_client, public_channel):
     ],
 )
 def test_validate_package_names_files_endpoint(
-    auth_client, public_channel, mocker, package_name, msg, config: Config
+    auth_client,
+    public_channel,
+    mocker,
+    package_name,
+    msg,
+    config: Config,
+    remove_package_versions,
 ):
 
     pkgstore = config.get_package_store()
@@ -508,7 +526,10 @@ def test_get_package_with_versions(
 
 @pytest.mark.parametrize("package_name", ["test-package"])
 def test_package_channel_data_attributes(
-    auth_client, make_package_version, channel_name
+    auth_client,
+    make_package_version,
+    channel_name,
+    remove_package_versions,
 ):
     # test attributes derived from channel data
     for package_filename in Path(".").glob("*.tar.bz2"):
diff --git a/quetz/tests/api/test_metrics.py b/quetz/tests/api/test_metrics.py
index 0ab80811..2ddd685e 100644
--- a/quetz/tests/api/test_metrics.py
+++ b/quetz/tests/api/test_metrics.py
@@ -132,12 +132,14 @@ def test_get_download_count(auth_client, public_channel, package_version, db, da
 
 
 @pytest.fixture
-def package_version_factory(channel_name, user, dao, public_package):
+def package_version_factory(channel_name, user, dao, public_package, db):
 
     package_format = "tarbz2"
     package_info = "{}"
     package_name = public_package.name
 
+    versions = []
+
     def factory(version, build_str=0, platform="linux-64"):
 
         filename = f"{package_name}-{version}-{build_str}.tar.bz2"
@@ -155,9 +157,14 @@ def factory(version, build_str=0, platform="linux-64"):
             user.id,
             size=11,
         )
+        versions.append(version)
         return version
 
-    return factory
+    yield factory
+
+    for version in versions:
+        db.delete(version)
+    db.commit()
 
 
 def test_get_channel_download_count(
diff --git a/quetz/tests/conftest.py b/quetz/tests/conftest.py
index b964a135..4d5f0535 100644
--- a/quetz/tests/conftest.py
+++ b/quetz/tests/conftest.py
@@ -32,6 +32,9 @@ def user_without_profile(db, user_role):
 
     yield new_user
 
+    db.delete(new_user)
+    db.commit()
+
 
 @fixture
 def user(db, user_without_profile):
diff --git a/quetz/tests/test_jobs.py b/quetz/tests/test_jobs.py
index 6824ecf4..995d234b 100644
--- a/quetz/tests/test_jobs.py
+++ b/quetz/tests/test_jobs.py
@@ -26,6 +26,16 @@
 pytest_plugins = ("pytest_asyncio",)
 
 
+@pytest.fixture
+def auto_rollback():
+    return False
+
+
+@pytest.fixture
+def sqlite_in_memory():
+    return False
+
+
 @pytest.fixture
 def package_name():
     return "my-package"
@@ -104,12 +114,15 @@ def package_role():
 
 
 @pytest.fixture
-def public_channel(dao: Dao, user, channel_role, channel_name):
+def public_channel(dao: Dao, user, channel_role, channel_name, db):
 
     channel_data = Channel(name=channel_name, private=False)
     channel = dao.create_channel(channel_data, user.id, channel_role)
 
-    return channel
+    yield channel
+
+    db.delete(channel)
+    db.commit()
 
 
 @pytest.fixture
@@ -125,11 +138,14 @@ def public_package(db, user, public_channel, dao, package_role, package_name):
 
 
 @pytest.fixture
-def manager(config):
+def manager(config, db):
 
     manager = SubprocessWorker("", {}, config)
     yield manager
+    manager._executor.shutdown()
     SubprocessWorker._executor = None
+    db.query(Job).delete()
+    db.commit()
 
 
 def test_create_task(db, user, package_version):
@@ -150,7 +166,7 @@ def failed_func(package_version: dict):
 
 
 def long_running(package_version: dict):
-    time.sleep(0.1)
+    time.sleep(0.25)
 
 
 def dummy_func(package_version: dict):
@@ -254,10 +270,14 @@ async def test_running_task(db, user, package_version, manager):
 
     assert task.status == TaskStatus.pending
 
-    time.sleep(0.01)
-    check_status(db)
+    # wait for task status to change
+    for i in range(50):
+        time.sleep(0.05)
+
+        db.refresh(task)
+        if task.status != TaskStatus.pending:
+            break
 
-    db.refresh(task)
     assert task.status == TaskStatus.running
 
     # wait for job to finish
@@ -286,8 +306,13 @@ async def test_restart_worker_process(db, user, package_version, manager, caplog
 
     assert task.status == TaskStatus.pending
 
-    time.sleep(0.01)
-    check_status(db)
+    # wait for task status to change
+    for i in range(50):
+        time.sleep(0.05)
+
+        db.refresh(task)
+        if task.status != TaskStatus.pending:
+            break
 
     db.refresh(task)
     assert task.status == TaskStatus.running
@@ -333,7 +358,7 @@ async def test_failed_task(db, user, package_version, manager):
     assert task.status == TaskStatus.failed
 
     db.refresh(job)
-    assert job.status == JobStatus.success
+    assert job.status == JobStatus.failed
 
 
 @pytest.mark.parametrize("items_spec", ["", None])
@@ -675,10 +700,11 @@ def test_validate_query_string(auth_client, query_str, ok):
         assert response.status_code == 422
 
 
-@pytest.mark.parametrize("user_role", ["owner"])
-@pytest.mark.parametrize("skip,limit", [(0, 2), (1, -1), (2, 1), (2, 5)])
-def test_jobs_pagination(auth_client, db, user, skip, limit):
+@pytest.fixture
+def many_jobs(db, user):
+
     n_jobs = 5
+    jobs = []
     for i in range(n_jobs):
         job = Job(
             id=i,
@@ -688,6 +714,20 @@ def test_jobs_pagination(auth_client, db, user, skip, limit):
             owner=user,
         )
         db.add(job)
+        jobs.append(job)
+
+    yield jobs
+
+    for job in jobs:
+        db.delete(job)
+    db.commit()
+
+
+@pytest.mark.parametrize("user_role", ["owner"])
+@pytest.mark.parametrize("skip,limit", [(0, 2), (1, -1), (2, 1), (2, 5)])
+def test_jobs_pagination(auth_client, skip, limit, many_jobs):
+
+    n_jobs = len(many_jobs)
 
     response = auth_client.get(f"/api/jobs?skip={skip}&limit={limit}")
     assert response.status_code == 200
@@ -735,13 +775,24 @@ def test_get_tasks(
         assert data['result'][0]['job_id'] == job.id
 
 
+@pytest.fixture()
+def other_user(db):
+
+    other_user = User(id=uuid.uuid4().bytes, username='otheruser')
+
+    db.add(other_user)
+
+    yield other_user
+
+    db.delete(other_user)
+    db.commit()
+
+
 @pytest.mark.parametrize("user_role", ["member"])
-def test_get_user_jobs(auth_client, db, user, package_version):
+def test_get_user_jobs(auth_client, db, user, package_version, other_user):
     job = Job(items_spec="*", owner=user, manifest=pickle.dumps(dummy_func))
     db.add(job)
 
-    other_user = User(id=uuid.uuid4().bytes, username='otheruser')
-
     other_job = Job(items_spec="*", owner=other_user, manifest=pickle.dumps(dummy_func))
     db.add(other_job)
 
diff --git a/quetz/tests/test_mirror.py b/quetz/tests/test_mirror.py
index cdb2811c..f1e21769 100644
--- a/quetz/tests/test_mirror.py
+++ b/quetz/tests/test_mirror.py
@@ -38,6 +38,14 @@ def proxy_channel(db):
     db.commit()
 
 
+@pytest.fixture(autouse=True)
+def remove_package_versions(db, user):
+    # we need to run this fixture before use, hence the dependency
+    # on user fixture
+    yield
+    db.query(PackageVersion).delete()
+
+
 @pytest.fixture
 def mirror_channel(dao, user, db):
 
diff --git a/quetz/tests/test_tasks.py b/quetz/tests/test_tasks.py
index 45ec34f2..23e0be58 100644
--- a/quetz/tests/test_tasks.py
+++ b/quetz/tests/test_tasks.py
@@ -2,6 +2,7 @@
 
 import pytest
 
+from quetz.db_models import PackageVersion
 from quetz.pkgstores import PackageStore
 from quetz.rest_models import Channel
 from quetz.tasks.indexing import validate_packages
@@ -42,6 +43,16 @@ def channel(dao, user, channel_name):
     return channel
 
 
+@pytest.fixture
+def remove_package_versions(db):
+    # clean up the created package versions after the test
+
+    yield
+
+    db.query(PackageVersion).delete()
+    db.commit()
+
+
 def test_reindex_package_files(
     config,
     user,
@@ -51,6 +62,7 @@ def test_reindex_package_files(
     dao,
     pkgstore: PackageStore,
     package_filenames,
+    remove_package_versions,
 ):
     user_id = user.id
     reindex_packages_from_store(dao, config, channel.name, user_id)
diff --git a/quetz/tests/test_versionorder.py b/quetz/tests/test_versionorder.py
index 7f8427ca..961d11db 100644
--- a/quetz/tests/test_versionorder.py
+++ b/quetz/tests/test_versionorder.py
@@ -226,7 +226,9 @@ def test_hexrd():
     assert sorted(vos) == vos
 
 
-def test_package_version(db, dao: Dao, user, channel_name, package_name):
+@pytest.fixture
+def package_with_versions(channel_name, package_name, dao, user, db):
+
     channel_data = Channel(name=channel_name, private=False)
     package_data = Package(name=package_name)
 
@@ -246,8 +248,9 @@ def test_package_version(db, dao: Dao, user, channel_name, package_name):
         ("0.1.0", 5),
         ("0.1.0", 2),
     ]
+    package_versions = []
     for ver, build_str in versions:
-        dao.create_version(
+        package_version = dao.create_version(
             channel_name,
             package_name,
             package_format,
@@ -260,7 +263,19 @@ def test_package_version(db, dao: Dao, user, channel_name, package_name):
             user.id,
             size=0,
         )
-    res = dao.get_package_versions(package)
+        package_versions.append(package_version)
+
+    yield package
+
+    for package_version in package_versions:
+        db.delete(package_version)
+        db.commit()
+
+
+def test_package_version(
+    db, dao: Dao, channel_name, package_name, package_with_versions
+):
+    res = dao.get_package_versions(package_with_versions)
     res_versions = [(VersionOrder(x[0].version), x[0].build_number) for x in res]
 
     assert sorted(res_versions, reverse=True) == res_versions