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

"""

 (c) 2015-2018 - Copyright Red Hat Inc

 Authors:

   Patrick Uiterwijk <puiterwijk@redhat.com></puiterwijk@redhat.com>

"""

from __future__ import unicode_literals, absolute_import

import datetime

import munch

import unittest

import shutil

import subprocess

import sys

import tempfile

import time

import os

import six

import json

import pygit2

import requests

from requests.adapters import HTTPAdapter

from requests.packages.urllib3.util.retry import Retry

from mock import patch, MagicMock

sys.path.insert(0, os.path.join(os.path.dirname(

    os.path.abspath(__file__)), '..'))

import pagure.lib.query

import pagure.cli.admin

import tests

REPOSPANNER_CONFIG_TEMPLATE = """

---

ca:

  path: %(path)s/repospanner/pki

admin:

  url:  https://nodea.regiona.repospanner.local:%(gitport)s/

  ca:   %(path)s/repospanner/pki/ca.crt

  cert: %(path)s/repospanner/pki/admin.crt

  key:  %(path)s/repospanner/pki/admin.key

storage:

  state: %(path)s/repospanner/state

  git:

    type: tree

    clustered: true

    directory: %(path)s/repospanner/git

listen:

  rpc:  127.0.0.1:%(rpcport)s

  http: 127.0.0.1:%(gitport)s

certificates:

  ca: %(path)s/repospanner/pki/ca.crt

  client:

    cert: %(path)s/repospanner/pki/nodea.regiona.crt

    key:  %(path)s/repospanner/pki/nodea.regiona.key

  server:

    default:

      cert: %(path)s/repospanner/pki/nodea.regiona.crt

      key:  %(path)s/repospanner/pki/nodea.regiona.key

hooks:

  bubblewrap:

    enabled: true

    unshare:

    - net

    - ipc

    - pid

    - uts

    share_net: false

    mount_proc: true

    mount_dev: true

    uid:

    gid:

    hostname: myhostname

    bind:

    ro_bind:

    - - /usr

      - /usr

    - - %(codepath)s

      - %(codepath)s

    - - %(path)s

      - %(path)s

    - - %(crosspath)s

      - %(crosspath)s

    symlink:

    - - usr/lib64

      - /lib64

    - - usr/bin

      - /bin

  runner: %(hookrunner_bin)s

  user: 0

"""

class PagureRepoSpannerTests(tests.Modeltests):

    """ Tests for repoSpanner integration of pagure """

    repospanner_binary = None

    repospanner_runlog = None

    repospanner_proc = None

    def run_cacmd(self, logfile, *args):

        """ Run a repoSpanner CA command. """

        subprocess.check_call(

            [self.repospanner_binary,

             '--config',

             os.path.join(self.path, 'repospanner', 'config.yml'),

             # NEVER use this in a production system! It makes repeatable keys

             'ca'] + list(args) + ['--very-insecure-weak-keys'],

            stdout=logfile,

            stderr=subprocess.STDOUT,

        )

    def setUp(self):

        """ set up the environment. """

        possible_paths = [

            './repospanner',

            '/usr/bin/repospanner',

        ]

        for option in possible_paths:

            option = os.path.abspath(option)

            if os.path.exists(option):

                self.repospanner_binary = option

                break

        if not self.repospanner_binary:

            raise unittest.SkipTest('repoSpanner not found')

        hookrunbin = os.path.join(os.path.dirname(self.repospanner_binary),

                                  'repohookrunner')

        if not os.path.exists(hookrunbin):

            raise Exception('repoSpanner found, but repohookrunner not')

        repobridgebin = os.path.join(os.path.dirname(self.repospanner_binary),

                                     'repobridge')

        if not os.path.exists(repobridgebin):

            raise Exception('repoSpanner found, but repobridge not')

        self.config_values['repobridge_binary'] = repobridgebin

        codepath = os.path.normpath(

            os.path.join(

                os.path.dirname(os.path.abspath(__file__)),

                "../"))

        # Only run the setUp() function if we are actually going ahead and run

        # this test. The reason being that otherwise, setUp will set up a

        # database, but because we "error out" from setUp, the tearDown()

        # function never gets called, leaving it behind.

        super(PagureRepoSpannerTests, self).setUp()

        # TODO: Find free ports

        configvals = {

            'path': self.path,

            'crosspath': tests.tests_state["path"],

            'gitport': 8443 + sys.version_info.major,

            'rpcport': 8445 + sys.version_info.major,

            'codepath': codepath,

            'hookrunner_bin': hookrunbin,

        }

        os.mkdir(os.path.join(self.path, 'repospanner'))

        cfgpath = os.path.join(self.path, 'repospanner', 'config.yml')

        with open(cfgpath, 'w') as cfg:

            cfg.write(REPOSPANNER_CONFIG_TEMPLATE % configvals)

        with open(os.path.join(self.path, 'repospanner', 'keylog'),

                  'w') as keylog:

            # Create the CA

            self.run_cacmd(keylog, 'init', 'repospanner.local')

            # Create the node cert

            self.run_cacmd(keylog, 'node', 'regiona', 'nodea')

            # Create the admin cert

            self.run_cacmd(keylog, 'leaf', 'admin',

                           '--admin', '--region', '*', '--repo', '*')

            # Create the Pagure cert

            self.run_cacmd(keylog, 'leaf', 'pagure',

                           '--read', '--write',

                           '--region', '*', '--repo', '*')

        with open(os.path.join(self.path, 'repospanner', 'spawnlog'),

                  'w') as spawnlog:

            # Initialize state

            subprocess.check_call(

                [self.repospanner_binary,

                 '--config', cfgpath,

                 'serve', '--spawn'],

                stdout=spawnlog,

                stderr=subprocess.STDOUT,

            )

        self.repospanner_runlog = open(

            os.path.join(self.path, 'repospanner', 'runlog'), 'w')

        try:

            self.repospanner_proc = subprocess.Popen(

                [self.repospanner_binary,

                 '--config', cfgpath,

                 'serve', '--debug'],

                stdout=self.repospanner_runlog,

                stderr=subprocess.STDOUT,

            )

            # Wait for the instance to become available

            resp = requests.get(

                'https://nodea.regiona.repospanner.local:%d/'

                % configvals['gitport'],

                verify=os.path.join(self.path, 'repospanner', 'pki', 'ca.crt'),

                cert=(

                    os.path.join(self.path, 'repospanner', 'pki', 'pagure.crt'),

                    os.path.join(self.path, 'repospanner', 'pki', 'pagure.key'),

                )

            )

            resp.raise_for_status()

            print('repoSpanner identification: %s' % resp.text)

        except:

            # Make sure to clean up repoSpanner, since we did start it

            self.tearDown()

            raise

    def tearDown(self):

        """ Tear down the repoSpanner instance. """

        if self.repospanner_proc:

            # Tear down

            self.repospanner_proc.terminate()

            exitcode = self.repospanner_proc.wait()

            if exitcode != 0:

                print('repoSpanner exit code: %d' % exitcode)

        if self.repospanner_runlog:

            self.repospanner_runlog.close()

        super(PagureRepoSpannerTests, self).tearDown()

class PagureRepoSpannerTestsNewRepoDefault(PagureRepoSpannerTests):

    config_values = {

        'repospanner_new_repo': "'default'",

        'authbackend': 'test_auth',

    }

    @patch('pagure.ui.app.admin_session_timedout')

    def test_new_project(self, ast):

        """ Test creating a new repo by default on repoSpanner works. """

        ast.return_value = False

        user = tests.FakeUser(username='foo')

        with tests.user_set(self.app.application, user):

            output = self.app.get('/new/')

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                'Create new Project', output_text)

            data = {

                'name': 'project-1',

                'description': 'Project #1',

                'create_readme': 'y',

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post('/new/', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nProject #1',

                output_text)

            self.assertIn(

                '<title>Overview - project-1 - Pagure</title>', output_text)

            self.assertIn('Added the README', output_text)

            output = self.app.get('/project-1/settings')

            self.assertIn(

                'This repository is on repoSpanner region default',

                output.get_data(as_text=True))

        with tests.user_set(self.app.application, tests.FakeUser(username='pingou')):

            data = {

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post(

                '/do_fork/project-1', data=data,

                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nProject #1',

                output_text)

            self.assertIn(

                '<title>Overview - project-1 - Pagure</title>', output_text)

            self.assertIn('Added the README', output_text)

            output = self.app.get('/fork/pingou/project-1/settings')

            self.assertIn(

                'This repository is on repoSpanner region default',

                output.get_data(as_text=True))

        # Verify that only pseudo repos exist, and no on-disk repos got created

        repodirlist = os.listdir(os.path.join(self.path, 'repos'))

        self.assertEqual(repodirlist, ['pseudo'])

    @patch.dict('pagure.config.config', {

        'ALLOW_HTTP_PULL_PUSH': True,

        'ALLOW_HTTP_PUSH': True,

        'HTTP_REPO_ACCESS_GITOLITE': False,

    })

    def test_http_pull(self):

        """ Test that the HTTP pull endpoint works for repoSpanner. """

        tests.create_projects(self.session)

        tests.create_tokens(self.session)

        tests.create_tokens_acl(self.session)

        self.create_project_full('clonetest', {"create_readme": "y"})

        # Verify the new project is indeed on repoSpanner

        project = pagure.lib.query._get_project(self.session, 'clonetest')

        self.assertTrue(project.is_on_repospanner)

        # Unfortunately, actually testing a git clone would need the app to

        # run on a TCP port, which the test environment doesn't do.

        output = self.app.get('/clonetest.git/info/refs?service=git-upload-pack')

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn("# service=git-upload-pack", output_text)

        self.assertIn("symref=HEAD:refs/heads/master", output_text)

        self.assertIn(" refs/heads/master\x00", output_text)

        output = self.app.post(

            '/clonetest.git/git-upload-pack',

            headers={'Content-Type': 'application/x-git-upload-pack-request'},

        )

        self.assertEqual(output.status_code, 400)

        output_text = output.get_data(as_text=True)

        self.assertIn("Error processing your request", output_text)

    @patch.dict('pagure.config.config', {

        'ALLOW_HTTP_PULL_PUSH': True,

        'ALLOW_HTTP_PUSH': True,

        'HTTP_REPO_ACCESS_GITOLITE': False,

    })

    def test_http_push(self):

        """ Test that the HTTP push endpoint works for repoSpanner. """

        tests.create_projects(self.session)

        tests.create_tokens(self.session)

        tests.create_tokens_acl(self.session)

        self.create_project_full('clonetest', {"create_readme": "y"})

        # Verify the new project is indeed on repoSpanner

        project = pagure.lib.query._get_project(self.session, 'clonetest')

        self.assertTrue(project.is_on_repospanner)

        # Unfortunately, actually testing a git clone would need the app to

        # run on a TCP port, which the test environment doesn't do.

        output = self.app.get(

            '/clonetest.git/info/refs?service=git-upload-pack',

            environ_overrides={'REMOTE_USER': 'pingou'},

        )

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn("# service=git-upload-pack", output_text)

        self.assertIn("symref=HEAD:refs/heads/master", output_text)

        self.assertIn(" refs/heads/master\x00", output_text)

    @patch('pagure.ui.app.admin_session_timedout')

    def test_hooks(self, ast):

        """ Test hook setting and running works. """

        ast.return_value = False

        pagure.cli.admin.session = self.session

        # Upload the hook script to repoSpanner

        args = munch.Munch({'region': 'default'})

        hookid = pagure.cli.admin.do_upload_repospanner_hooks(args)

        user = tests.FakeUser(username='foo')

        with tests.user_set(self.app.application, user):

            data = {

                'name': 'project-1',

                'description': 'Project #1',

                'create_readme': 'y',

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post('/new/', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nProject #1',

                output_text)

            self.assertIn(

                '<title>Overview - project-1 - Pagure</title>', output_text)

            self.assertIn('Added the README', output_text)

            output = self.app.get('/project-1/settings')

            self.assertIn(

                'This repository is on repoSpanner region default',

                output.get_data(as_text=True))

            # Check file before the commit:

            output = self.app.get('/project-1/raw/master/f/README.md')

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertEqual(output_text, '# project-1\n\nProject #1')

        # Set the hook

        args = munch.Munch({'hook': hookid})

        projects = pagure.cli.admin.do_ensure_project_hooks(args)

        self.assertEqual(["project-1"], projects)

        with tests.user_set(self.app.application, user):

            # Set editing Denied

            self.set_auth_status(False)

            # Try to make an edit in the repo

            data = {

                'content': 'foo\n bar\n  baz',

                'commit_title': 'test commit',

                'commit_message': 'Online commit',

                'email': 'foo@bar.com',

                'branch': 'master',

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post(

                '/project-1/edit/master/f/README.md', data=data,

                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                "Remote hook declined the push: ",

                output_text

            )

            self.assertIn(

                "Denied push for ref 'refs/heads/master' for user 'foo'\n"

                "All changes have been rejected",

                output_text

            )

            # Check file after the commit:

            output = self.app.get('/project-1/raw/master/f/README.md')

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertEqual(output_text, '# project-1\n\nProject #1')

            # Set editing Allowed

            self.set_auth_status(True)

            # Try to make an edit in the repo

            data = {

                'content': 'foo\n bar\n  baz',

                'commit_title': 'test commit',

                'commit_message': 'Online commit',

                'email': 'foo@bar.com',

                'branch': 'master',

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post(

                '/project-1/edit/master/f/README.md', data=data,

                follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '<title>Commits - project-1 - Pagure</title>', output_text)

            # Check file after the commit:

            output = self.app.get('/project-1/raw/master/f/README.md')

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertEqual(output_text, 'foo\n bar\n  baz')

    @patch.dict('pagure.config.config', {'PAGURE_ADMIN_USERS': ['pingou'],

                                         'ALLOW_ADMIN_IGNORE_EXISTING_REPOS': True})

    @patch('pagure.ui.app.admin_session_timedout')

    def test_adopt_project(self, ast):

        """ Test adopting a project in repoSpanner works. """

        ast.return_value = False

        user = tests.FakeUser(username='foo')

        with tests.user_set(self.app.application, user):

            output = self.app.get('/new/')

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                'Create new Project', output_text)

            data = {

                'name': 'project-1',

                'description': 'Project #1',

                'create_readme': 'y',

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post('/new/', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nProject #1',

                output_text)

            self.assertIn(

                '<title>Overview - project-1 - Pagure</title>', output_text)

            self.assertIn('Added the README', output_text)

            output = self.app.get('/project-1/settings')

            self.assertIn(

                'This repository is on repoSpanner region default',

                output.get_data(as_text=True))

        # Delete the project instance so that the actual repo remains

        project = pagure.lib.query._get_project(self.session, 'project-1')

        self.session.delete(project)

        self.session.commit()

        shutil.rmtree(os.path.join(self.path, 'repos', 'pseudo'))

        user = tests.FakeUser(username='pingou')

        with tests.user_set(self.app.application, user):

            output = self.app.get('/project-1/')

            self.assertEqual(output.status_code, 404)

            data = {

                'name': 'project-1',

                'description': 'Recreated project #1',

                'create_readme': 'false',

                'csrf_token': self.get_csrf(),

            }

            output = self.app.post('/new/', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                'Repo pagure/main/project-1 already exists',

                output_text)

            data['ignore_existing_repos'] = 'y'

            output = self.app.post('/new/', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nRecreated project #1',

                output_text)

            self.assertIn(

                '<title>Overview - project-1 - Pagure</title>', output_text)

            self.assertIn('Added the README', output_text)

if __name__ == '__main__':

    unittest.main(verbosity=2)