torc 0.21.0

Workflow management system
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

"""Test workflow execution"""

import json
import os
import subprocess
import time
from datetime import datetime
from pathlib import Path

import pytest
from click.testing import CliRunner
from torc.openapi_client.api.default_api import DefaultApi
from torc.openapi_client.models.user_data_model import UserDataModel
from torc.openapi_client.models.compute_node_model import ComputeNodeModel
from torc.openapi_client.models.compute_nodes_resources import ComputeNodesResources
from torc.openapi_client.models.result_model import ResultModel
from torc.openapi_client.models.job_model import JobModel
from torc.api import iter_documents, create_jobs, wait_for_healthy_database
from torc.common import GiB
from torc.exceptions import InvalidWorkflow
from torc.common import timer_stats_collector


def test_add_bulk_jobs(mapped_function_workflow):
    """Test the add_jobs helper function."""
    db, _ = mapped_function_workflow
    api = db.api
    initial_jobs = api.list_jobs(db.workflow.id).items
    initial_count = len(initial_jobs)
    resource_requirements = api.list_resource_requirements(db.workflow.id).items[0]

    jobs = [
        JobModel(
            workflow_id=db.workflow.id,
            name=f"added_job{i}",
            command="python my_script.py",
            resource_requirements_id=resource_requirements.id,
        )
        for i in range(1, 51)
    ]

    added_jobs = create_jobs(api, jobs, max_transfer_size=11)
    assert len(added_jobs) == 50

    final_jobs = api.list_jobs(db.workflow.id).items
    assert len(final_jobs) == initial_count + 50
    added_names = {x.name for x in final_jobs if x.name.startswith("added_job")}
    expected_names = {f"added_job{i}" for i in range(1, 51)}
    assert added_names == expected_names