Functional tests for RepositoryVersions

vrtest.py

import time
import unittest

import requests
from requests.auth import HTTPBasicAuth


# usage: python3 -m unittest vrtest.py


api = 'http://pulp3.dev:8000/api/v3/'
auth = HTTPBasicAuth('admin', 'admin')
repo_name = 'foo'
importer_name = 'foo_imp'
publisher_name = 'foo_pub'
distributor_name = 'foo_dist'
feed_base = 'http://fedorapeople.org/~mhrivnak/vrtests/'
importer_data = {
        'name': importer_name,
        'download_policy': 'immediate',
        'sync_mode': 'mirror',
        'validate': True,
        'ssl_validation': True,
        'feed_url': feed_base + '1/PULP_MANIFEST',
}
publisher_data = {
    'name': publisher_name,
}
distribution_data = {
    'name': distributor_name,
    'base_path': 'foo',
    'http': True,
    'auto_updated': True,
    'publication': '',
}


def poll_task(response):
    task_urls = [task['_href'] for task in response.json()]
    for url in task_urls:
        while True:
            status = requests.get(url, auth=auth).json()
            if status['state'] == 'waiting':
                print(status['state'])
            else:
                for report in status['progress_reports']:
                    del report['task']
                    del report['suffix']
                    print(report)
            if status['state'] in ['completed', 'failed']:
                print(status['state'])
                break
            time.sleep(.3)


def delete_repo(name):
    r = requests.get(api + 'repositories/?name=' + repo_name, auth=auth)
    if r.json()['results']:
        repo_url = r.json()['results'][0]['_href']
        print('deleting repo: ' + repo_url)
        r = requests.delete(repo_url, auth=auth)
        poll_task(r)


def delete_distribution(base_path):
    r = requests.get(api + 'distributions/?base_path=' + base_path, auth=auth)
    if r.json()['results']:
        dist_url = r.json()['results'][0]['_href']
        print('deleting distribution: ' + dist_url)
        r = requests.delete(dist_url, auth=auth)


def find_version(versions_href, number):
    versions = requests.get(versions_href, auth=auth).json()
    for v in versions['results']:
        if v['number'] == number:
            return v
            break


class TestRepoVersions(unittest.TestCase):
    """
    These tests simulate changes in a remote repository by changing the
    feed_url for the repo between sync tasks.

    They leave behind the test data in case you want data to browse and play
    with. The next run will delete old test data and start fresh.
    """
    @classmethod
    def setUpClass(cls):
        # remove data from any previous test run
        delete_repo(repo_name)
        delete_distribution(distribution_data['base_path'])

        r = requests.post(api + 'repositories/', auth=auth, data={'name': repo_name})
        cls.repo_url = r.json()['_href']
        importer_data['repository'] = cls.repo_url
        publisher_data['repository'] = cls.repo_url
        print('added repository: ' + cls.repo_url)

        r = requests.post(api+ 'importers/file/', auth=auth, data=importer_data)
        cls.importer_url = r.json()['_href']
        print('added importer: ' + cls.importer_url)

        r = requests.post(api + 'publishers/file/', auth=auth, data=publisher_data)
        cls.publisher_url = r.json()['_href']
        distribution_data['publisher'] = cls.publisher_url
        print('added publisher: ' + cls.publisher_url)

        r = requests.post(api + 'distributions/', auth=auth, data=distribution_data)
        cls.distribution_url = r.json()['_href']
        print('added distribution: ' + cls.distribution_url)

    def assertFilesInVersion(self, filenames, version_number):
        # get the version
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        version = find_version(versions_href, version_number)
        self.assertTrue(version is not None)

        content_href = version['_content_href']
        content = requests.get(content_href, auth=auth).json()
        found_files = {c['path'] for c in content['results']}
        self.assertEqual(filenames, found_files)

    def update_feed(self, url):
        importer = requests.get(self.importer_url, auth=auth).json()
        importer['feed_url'] = url

        r = requests.put(self.importer_url, auth=auth, data=importer)
        poll_task(r)

    def test_1_first_sync(self):
        # no versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 0)

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # one version exists
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 1)
        self.assertEqual(versions['results'][0]['number'], 1)

        self.assertFilesInVersion({'a.bin', 'b.bin'}, 1)

    def test_2_adds_c(self):
        self.update_feed(feed_base + '2/PULP_MANIFEST')

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # two versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 2)

        self.assertFilesInVersion({'a.bin', 'b.bin', 'c.bin'}, 2)

    def test_3_removed_a(self):
        self.update_feed(feed_base + '3/PULP_MANIFEST')

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # three versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 3)

        self.assertFilesInVersion({'b.bin', 'c.bin'}, 3)

    def test_4_add_and_remove(self):
        self.update_feed(feed_base + '4/PULP_MANIFEST')

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # four versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 4)

        self.assertFilesInVersion({'c.bin', 'd.bin'}, 4)

    def test_5_add_secret(self):
        self.update_feed(feed_base + '5/PULP_MANIFEST')

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # five versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 5)

        self.assertFilesInVersion({'c.bin', 'd.bin', 'e.bin'}, 5)

    def test_6_remove_secret(self):
        self.update_feed(feed_base + '6/PULP_MANIFEST')

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # six versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 6)

        self.assertFilesInVersion({'c.bin', 'd.bin'}, 6)

        # delete version five
        v5 = find_version(versions_href, 5)
        r = requests.delete(v5['_href'], auth=auth)
        poll_task(r)

        # five versions remain
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 5)

        # version six content unchanged
        self.assertFilesInVersion({'c.bin', 'd.bin'}, 6)

    def test_7_add_remove_add(self):
        self.update_feed(feed_base + '7/PULP_MANIFEST')

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # six versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 6)

        self.assertFilesInVersion({'c.bin', 'd.bin', 'f.bin'}, 7)

        # remove f
        self.update_feed(feed_base + '6/PULP_MANIFEST')
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # add f back
        self.update_feed(feed_base + '7/PULP_MANIFEST')
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # eight versions exist
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 8)

        self.assertFilesInVersion({'c.bin', 'd.bin'}, 8)
        self.assertFilesInVersion({'c.bin', 'd.bin', 'f.bin'}, 9)

        # delete version 8, which results is a potentially weird scenario.
        # Either there will be two different RepositoryContent records showing
        # f.bin removed in 9, and the next one adding it in 9. Or the code will
        # be smarter and reduce those back to one RepositoryContent record.
        v8 = find_version(versions_href, 8)
        r = requests.delete(v8['_href'], auth=auth)
        poll_task(r)

        # make sure version 8 is gone
        r = requests.get(v8['_href'], auth=auth)
        self.assertEqual(r.status_code, 404)

        # ensure that versions 7 and 9 report the correct content
        self.assertFilesInVersion({'c.bin', 'd.bin', 'f.bin'}, 7)
        self.assertFilesInVersion({'c.bin', 'd.bin', 'f.bin'}, 9)


    def test_no_change(self):
        """
        when nothing has changed in the remote repo, make sure a new version
        does not get created.
        """
        # seven versions exist
        repo = requests.get(self.repo_url, auth=auth).json()
        versions_href = repo['_versions_href']
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 7)

        # perform a sync
        r = requests.post(self.importer_url + 'sync/', auth=auth)
        poll_task(r)

        # still seven versions
        versions = requests.get(versions_href, auth=auth).json()
        self.assertEqual(len(versions['results']), 7)

    def test_cannot_delete_latest(self):
        """
        The latest version cannot be deleted, because then a new version could
        later get created with the same number. That would violate the
        guarantee that a unique version's content set is immutable.
        """
        # try to delete the latest version
        repo = requests.get(self.repo_url, auth=auth).json()
        latest_href = repo['_latest_version_href']
        r = requests.delete(latest_href, auth=auth)
        poll_task(r)

        # make sure it still exists
        r = requests.get(latest_href, auth=auth)
        self.assertEqual(r.status_code, 200)