Repository: ansible/awx
File: awx/main/tests/unit/models/test_jobs.py (Python)

Code before:
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=0)

    for host in hosts:
        filepath = os.path.join(fact_cache, host.name)
        assert os.path.exists(filepath)
        with open(filepath, 'r') as f:
            assert f.read() == json.dumps(host.ansible_facts)
        assert os.path.getmtime(filepath) <= last_modified


@pytest.mark.django_db
def test_fact_cache_with_invalid_path_traversal(tmpdir):
    hosts = [
        Host(
            name='../foo',
            ansible_facts={"a": 1, "b": 2},
        ),
    ]

    fact_cache = os.path.join(tmpdir, 'facts')
    start_fact_cache(hosts, fact_cache, timeout=0)
    # a file called "foo" should _not_ be written outside the facts dir
    assert os.listdir(os.path.join(fact_cache, '..')) == ['facts']


@pytest.mark.django_db
def test_start_job_fact_cache_past_timeout(hosts, tmpdir):
    fact_cache = os.path.join(tmpdir, 'facts')
    # the hosts fixture was modified 5s ago, which is more than 2s
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=2)
    assert last_modified is None

    for host in hosts:
        assert not os.path.exists(os.path.join(fact_cache, host.name))


@pytest.mark.django_db
def test_start_job_fact_cache_within_timeout(hosts, tmpdir):
    fact_cache = os.path.join(tmpdir, 'facts')
    # the hosts fixture was modified 5s ago, which is less than 7s
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=7)
    assert last_modified

    for host in hosts:
        assert os.path.exists(os.path.join(fact_cache, host.name))


@pytest.mark.django_db
def test_finish_job_fact_cache_with_existing_data(hosts, mocker, tmpdir, ref_time):
    fact_cache = os.path.join(str(tmpdir), 'facts')
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=0)

Review comment (suggestion):
We shouldn't use this for any tests under `awx/main/tests/unit`. That's the idea of the separation of folders. If the test doesn't need the database, don't add the marker. If it does, then move to `awx/main/tests/functional`.

Code after:
        assert os.path.exists(filepath)
        with open(filepath, 'r') as f:
            assert f.read() == json.dumps(host.ansible_facts)
        assert os.path.getmtime(filepath) <= last_modified


def test_fact_cache_with_invalid_path_traversal(tmpdir):
    hosts = [
        Host(
            name='../foo',
            ansible_facts={"a": 1, "b": 2},
        ),
    ]

    fact_cache = os.path.join(tmpdir, 'facts')
    start_fact_cache(hosts, fact_cache, timeout=0)
    # a file called "foo" should _not_ be written outside the facts dir
    assert os.listdir(os.path.join(fact_cache, '..')) == ['facts']


def test_start_job_fact_cache_past_timeout(hosts, tmpdir):
    fact_cache = os.path.join(tmpdir, 'facts')
    # the hosts fixture was modified 5s ago, which is more than 2s
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=2)
    assert last_modified is None

    for host in hosts:
        assert not os.path.exists(os.path.join(fact_cache, host.name))


def test_start_job_fact_cache_within_timeout(hosts, tmpdir):
    fact_cache = os.path.join(tmpdir, 'facts')
    # the hosts fixture was modified 5s ago, which is less than 7s
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=7)
    assert last_modified

    for host in hosts:
        assert os.path.exists(os.path.join(fact_cache, host.name))


def test_finish_job_fact_cache_clear(hosts, mocker, ref_time, tmpdir):
    fact_cache = os.path.join(tmpdir, 'facts')
    last_modified, _ = start_fact_cache(hosts, fact_cache, timeout=0)

    bulk_update = mocker.patch('awx.main.tasks.facts.bulk_update_sorted_by_id')
    mocker.patch('os.path.exists', side_effect=lambda path: hosts[1].name not in path)

    # Simulate one host's fact file getting deleted
    os.remove(os.path.join(fact_cache, hosts[1].name))
    finish_fact_cache(hosts, fact_cache, last_modified)