# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import json
import shutil
import os
import tg
import pkg_resources
from pylons import tmpl_context as c
from ming.orm import ThreadLocalORMSession
from mock import patch
from nose.tools import assert_equal
from IPython.testing.decorators import onlyif
from allura import model as M
from allura.lib import helpers as h
from alluratest.controller import TestController
from forgesvn.tests import with_svn
from allura.tests.decorators import with_tool
class SVNTestController(TestController):
def setUp(self):
TestController.setUp(self)
self.setup_with_tools()
def _make_app(self, mount_point, name):
h.set_context('test', mount_point, neighborhood='Projects')
repo_dir = pkg_resources.resource_filename(
'forgesvn', 'tests/data/')
c.app.repo.fs_path = repo_dir
c.app.repo.status = 'ready'
c.app.repo.name = name
c.app.repo.refresh()
if os.path.isdir(c.app.repo.tarball_path):
shutil.rmtree(c.app.repo.tarball_path)
ThreadLocalORMSession.flush_all()
ThreadLocalORMSession.close_all()
@with_svn
@with_tool('test', 'SVN', 'svn-tags', 'SVN with tags')
def setup_with_tools(self):
self._make_app('svn-tags', 'testsvn-trunk-tags-branches')
self._make_app('src', 'testsvn')
class TestRootController(SVNTestController):
def test_status(self):
resp = self.app.get('/src/status')
d = json.loads(resp.body)
assert d == dict(status='ready')
def test_status_html(self):
resp = self.app.get('/src/').follow()
# repo status not displayed if 'ready'
assert None == resp.html.find('div', dict(id='repo_status'))
h.set_context('test', 'src', neighborhood='Projects')
c.app.repo.status = 'analyzing'
ThreadLocalORMSession.flush_all()
ThreadLocalORMSession.close_all()
# repo status displayed if not 'ready'
resp = self.app.get('/src/').follow()
div = resp.html.find('div', dict(id='repo_status'))
assert div.span.text == 'analyzing'
def test_index(self):
resp = self.app.get('/src/').follow()
assert 'svn checkout' in resp
assert '[r5]' in resp, resp.showbrowser()
def test_index_empty(self):
self.app.get('/svn/')
def test_commit_browser(self):
resp = self.app.get('/src/commit_browser')
def test_commit_browser_data(self):
resp = self.app.get('/src/commit_browser_data')
data = json.loads(resp.body);
assert data['max_row'] == 4
assert data['next_column'] == 1
for val in data['built_tree'].values():
if val['url'] == '/p/test/src/1/':
assert val['short_id'] == '[r1]'
assert val['column'] == 0
assert val['row'] == 4
assert val['message'] == 'Create readme'
def test_feed(self):
for ext in ['', '.rss']:
r = self.app.get('/src/feed%s' % ext)
channel = r.xml.find('channel')
title = channel.find('title').text
assert_equal(title, 'test SVN changes')
description = channel.find('description').text
assert_equal(description, 'Recent changes to SVN repository in test project')
link = channel.find('link').text
assert_equal(link, 'http://localhost:80/p/test/src/')
commit = channel.find('item')
assert_equal(commit.find('title').text, 'Create readme')
link = 'http://localhost:80/p/test/src/1/'
assert_equal(commit.find('link').text, link)
assert_equal(commit.find('guid').text, link)
# .atom has slightly different structure
prefix = '{http://www.w3.org/2005/Atom}'
r = self.app.get('/src/feed.atom')
title = r.xml.find(prefix + 'title').text
assert_equal(title, 'test SVN changes')
link = r.xml.find(prefix + 'link').attrib['href']
assert_equal(link, 'http://localhost:80/p/test/src/')
commit = r.xml.find(prefix + 'entry')
assert_equal(commit.find(prefix + 'title').text, 'Create readme')
link = 'http://localhost:80/p/test/src/1/'
assert_equal(commit.find(prefix + 'link').attrib['href'], link)
def test_commit(self):
resp = self.app.get('/src/3/tree/')
assert len(resp.html.findAll('tr')) == 3, resp.showbrowser()
def test_tree(self):
resp = self.app.get('/src/1/tree/')
assert len(resp.html.findAll('tr')) == 2, resp.showbrowser()
resp = self.app.get('/src/3/tree/a/')
assert len(resp.html.findAll('tr')) == 2, resp.showbrowser()
def test_file(self):
resp = self.app.get('/src/1/tree/README')
assert 'README' in resp.html.find('h2', {'class':'dark title'}).contents[2]
content = str(resp.html.find('div', {'class':'clip grid-19 codebrowser'}))
assert 'This is readme' in content, content
assert '<span id="l1" class="code_block">' in resp
assert 'var hash = window.location.hash.substring(1);' in resp
def test_invalid_file(self):
resp = self.app.get('/src/1/tree/READMEz', status=404)
def test_diff(self):
resp = self.app.get('/src/3/tree/README?diff=2')
assert 'This is readme' in resp, resp.showbrowser()
assert '+++' in resp, resp.showbrowser()
def test_checkout_svn(self):
self.app.post('/p/test/admin/src/set_checkout_url',
{"checkout_url": "badurl"})
r = self.app.get('/p/test/admin/src/checkout_url')
assert 'value="badurl"' not in r
self.app.post('/p/test/admin/src/set_checkout_url',
{"checkout_url": ""})
r = self.app.get('/p/test/admin/src/checkout_url')
assert 'value="trunk"' not in r
self.app.post('/p/test/admin/src/set_checkout_url',
{"checkout_url": "a"})
r = self.app.get('/p/test/admin/src/checkout_url')
assert 'value="a"' in r
def test_log(self):
r = self.app.get('/src/1/log/')
assert 'Create readme' in r
r = self.app.get('/src/2/log/?path=')
assert "Create readme" in r
assert "Add path" in r
r = self.app.get('/src/2/log/?path=README')
assert "Modify readme" not in r
assert "Create readme" in r
r = self.app.get('/src/2/log/?path=/a/b/c/')
assert 'Add path' in r
assert 'Remove hello.txt' not in r
r = self.app.get('/src/5/log/?path=a/b/c/')
assert 'Add path' in r
assert 'Remove hello.txt' in r
r = self.app.get('/src/2/log/?path=does/not/exist/')
assert 'No (more) commits' in r
@onlyif(os.path.exists(tg.config.get('scm.repos.tarball.zip_binary', '/usr/bin/zip')), 'zip binary is missing')
def test_tarball(self):
r = self.app.get('/src/3/tree/')
assert 'Download Snapshot' in r
r = self.app.post('/src/3/tarball').follow()
assert 'Checking snapshot status...' in r
r = self.app.get('/src/3/tarball')
assert 'Checking snapshot status...' in r
M.MonQTask.run_ready()
ThreadLocalORMSession.flush_all()
r = self.app.get('/src/3/tarball_status')
assert '{"status": "ready"}' in r
r = self.app.get('/src/3/tarball')
assert 'Your download will begin shortly' in r
@onlyif(os.path.exists(tg.config.get('scm.repos.tarball.zip_binary', '/usr/bin/zip')), 'zip binary is missing')
def test_tarball_tags_aware(self):
h.set_context('test', 'svn-tags', neighborhood='Projects')
shutil.rmtree(c.app.repo.tarball_path, ignore_errors=True)
r = self.app.get('/p/test/svn-tags/19/tree/')
form = r.html.find('form', 'tarball')
assert_equal(form.button.text, 'Download Snapshot')
assert_equal(form.get('action'), '/p/test/svn-tags/19/tarball')
r = self.app.get('/p/test/svn-tags/19/tree/tags/tag-1.0/')
form = r.html.find('form', 'tarball')
assert_equal(form.button.text, 'Download Snapshot')
assert_equal(form.get('action'), '/p/test/svn-tags/19/tarball')
assert_equal(form.find('input', attrs=dict(name='path')).get('value'), '/tags/tag-1.0')
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/tags/tag-1.0')
assert_equal(r.json['status'], 'na')
r = self.app.post('/p/test/svn-tags/19/tarball', dict(path='/tags/tag-1.0')).follow()
assert 'Checking snapshot status...' in r
M.MonQTask.run_ready()
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/tags/tag-1.0')
assert_equal(r.json['status'], 'ready')
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/trunk')
assert_equal(r.json['status'], 'na')
r = self.app.post('/p/test/svn-tags/19/tarball', dict(path='/trunk/')).follow()
assert 'Checking snapshot status...' in r
M.MonQTask.run_ready()
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/trunk')
assert_equal(r.json['status'], 'ready')
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/branches/aaa/')
assert_equal(r.json['status'], 'na')
# All of the following also should be ready because...
# ...this is essentially the same as trunk snapshot
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/trunk/some/path/')
assert_equal(r.json['status'], 'ready')
r = self.app.get('/p/test/svn-tags/19/tarball_status')
assert_equal(r.json['status'], 'ready')
# ...the same as trunk, 'cause concrete tag isn't specified
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/tags/')
assert_equal(r.json['status'], 'ready')
# ...the same as trunk, 'cause concrete branch isn't specified
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/branches/')
assert_equal(r.json['status'], 'ready')
# ...this is essentially the same as tag snapshot
r = self.app.get('/p/test/svn-tags/19/tarball_status?path=/tags/tag-1.0/dir')
assert_equal(r.json['status'], 'ready')
class TestImportController(SVNTestController):
def test_index(self):
r = self.app.get('/p/test/admin/src/importer').follow(status=200)
assert 'You cannot import into a repository that already has commits in it.' in r
@patch('forgesvn.svn_main.allura.tasks.repo_tasks')
def test_do_import(self, tasks):
r = self.app.post('/p/test/admin/src/importer/do_import',
{'checkout_url': 'http://fake.svn/'})
assert not tasks.reclone.post.called
@with_tool('test', 'SVN', 'empty', 'empty SVN')
def test_index_empty_repo(self):
r = self.app.get('/p/test/admin/empty/importer').follow(status=200)
assert 'Enter the URL of the source repository below' in r
@patch('forgesvn.svn_main.allura.tasks.repo_tasks')
@with_tool('test', 'SVN', 'empty', 'empty SVN')
def test_do_import_empty_repo(self, tasks):
r = self.app.post('/p/test/admin/empty/importer/do_import',
{'checkout_url': 'http://fake.svn/'})
assert tasks.reclone.post.called