pagure/tests/test_pagure_flask_ui_archives.py

# -*- coding: utf-8 -*-

"""
 (c) 2018 - Copyright Red Hat Inc

 Authors:
   Clement Verna <cverna@tutanota.com>

"""

from __future__ import unicode_literals

import unittest
import sys
import os
import time

import mock
import pygit2

sys.path.insert(0, os.path.join(os.path.dirname(
    os.path.abspath(__file__)), '..'))

import pagure.lib.git
import pagure.lib.query
import tests
from tests.test_pagure_lib_git_get_tags_objects import add_repo_tag


class PagureFlaskUiArchivesTest(tests.Modeltests):
    """ Tests checking the archiving mechanism. """


    def setUp(self):
        """ Set up the environnment, ran before every tests. """
        super(PagureFlaskUiArchivesTest, self).setUp()
        tests.create_projects(self.session)
        tests.create_projects_git(os.path.join(self.path, 'repos'), bare=True)
        project = pagure.lib.query._get_project(self.session, 'test')

        # test has both commits and tags
        repopath = os.path.join(self.path, 'repos', 'test.git')
        tests.add_readme_git_repo(repopath)
        repo = pygit2.Repository(repopath)
        add_repo_tag(self.path, repo, ['v1.0', 'v1.1'], 'test.git')

        # test2 has only commits
        tests.add_readme_git_repo(os.path.join(
            self.path, 'repos', 'test2.git'))

        # somenamespace/test3 has neither commits nor tags

        # Create the archive folder:
        self.archive_path = os.path.join(self.path, 'archives')
        os.mkdir(self.archive_path)

    def test_project_no_conf(self):
        """ Test getting the archive when pagure isn't configured. """
        output = self.app.get(
            '/somenamespace/test3/archive/tag1/test3-tag1.zip',
            follow_redirects=True)

        self.assertEqual(output.status_code, 404)
        self.assertIn(
            "This pagure instance isn&#39;t configured to support "
            "this feature", output.get_data(as_text=True))

        self.assertEqual(os.listdir(self.archive_path), [])

    def test_project_invalid_conf(self):
        """ Test getting the archive when pagure is wrongly configured. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'invalid')}):
            output = self.app.get(
                '/somenamespace/test3/archive/tag1/test3-tag1.zip',
                follow_redirects=True)

            self.assertEqual(output.status_code, 500)
            self.assertIn(
                "Incorrect configuration, please contact your admin",
                output.get_data(as_text=True))

        self.assertEqual(os.listdir(self.archive_path), [])

    def test_project_invalid_format(self):
        """ Test getting the archive when the format provided is invalid. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/somenamespace/test3/archive/tag1/test3-tag1.unzip',
                follow_redirects=True)

            self.assertEqual(output.status_code, 404)

        self.assertEqual(os.listdir(self.archive_path), [])

    def test_project_no_commit(self):
        """ Test getting the archive of an empty project. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/somenamespace/test3/archive/tag1/test3-tag1.zip',
                follow_redirects=True)

            self.assertEqual(output.status_code, 404)
            self.assertIn(
                "<p>Invalid commit provided</p>",
                output.get_data(as_text=True))

        self.assertEqual(os.listdir(self.archive_path), [])

    def test_project_no_tag(self):
        """ Test getting the archive of a non-empty project but without
        tags. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test2/archive/tag1/test2-tag1.zip',
                follow_redirects=True)

            self.assertEqual(output.status_code, 404)
            self.assertIn(
                "<p>Invalid commit provided</p>",
                output.get_data(as_text=True))

        self.assertEqual(os.listdir(self.archive_path), [])

    def test_project_no_tag(self):
        """ Test getting the archive of an empty project. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test2/archive/tag1/test2-tag1.zip',
                follow_redirects=True)

            self.assertEqual(output.status_code, 404)
            self.assertIn(
                "<p>Invalid commit provided</p>",
                output.get_data(as_text=True))

        self.assertEqual(os.listdir(self.archive_path), [])

    def test_project_w_tag_zip(self):
        """ Test getting the archive from a tag. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test/archive/v1.0/test-v1.0.zip',
                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

        self.assertEqual(
            os.listdir(self.archive_path), ['test'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test')),
            ['tags'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test', 'tags')),
            ['v1.0'])

        self.assertEqual(
            len(os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0'))),
            1)

        files = os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0'))
        self.assertEqual(
            os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0', files[0])),
            ['test-v1.0.zip'])

    def test_project_w_tag_tar(self):
        """ Test getting the archive from a tag. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test/archive/v1.0/test-v1.0.tar',
                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

        self.assertEqual(
            os.listdir(self.archive_path), ['test'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test')),
            ['tags'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test', 'tags')),
            ['v1.0'])

        self.assertEqual(
            len(os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0'))),
            1)

        files = os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0'))
        self.assertEqual(
            os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0', files[0])),
            ['test-v1.0.tar'])

    def test_project_w_tag_tar_gz(self):
        """ Test getting the archive from a tag. """
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test/archive/v1.0/test-v1.0.tar.gz',
                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

        self.assertEqual(
            os.listdir(self.archive_path), ['test'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test')),
            ['tags'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test', 'tags')),
            ['v1.0'])

        self.assertEqual(
            len(os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0'))),
            1)

        files = os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0'))
        self.assertEqual(
            os.listdir(os.path.join(
                self.archive_path, 'test', 'tags', 'v1.0', files[0])),
            ['test-v1.0.tar.gz'])

    def test_project_w_commit_tar_gz(self):
        """ Test getting the archive from a commit. """
        repopath = os.path.join(self.path, 'repos', 'test.git')
        repo = pygit2.Repository(repopath)
        commit = repo.head.target.hex
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test/archive/%s/test-v1.0.tar.gz' % commit,
                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

        self.assertEqual(
            os.listdir(self.archive_path), ['test'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test')),
            [commit])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test', commit)),
            ['test-v1.0.tar.gz'])

    def test_project_w_commit_tar_gz_twice(self):
        """ Test getting the archive from a commit twice, so we hit the
        disk cache. """
        repopath = os.path.join(self.path, 'repos', 'test.git')
        repo = pygit2.Repository(repopath)
        commit = repo.head.target.hex
        with mock.patch.dict(
                'pagure.config.config',
                {'ARCHIVE_FOLDER': os.path.join(self.path, 'archives')}):
            output = self.app.get(
                '/test/archive/%s/test-v1.0.tar.gz' % commit,
                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output = self.app.get(
                '/test/archive/%s/test-v1.0.tar.gz' % commit,
                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

        self.assertEqual(
            os.listdir(self.archive_path), ['test'])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test')),
            [commit])
        self.assertEqual(
            os.listdir(os.path.join(self.archive_path, 'test', commit)),
            ['test-v1.0.tar.gz'])


if __name__ == '__main__':
    unittest.main(verbosity=2)