ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Merge branch 'devel' of github.com:ansible/ansible into devel

Browse source
This commit is contained in:
Michael DeHaan 2012-08-24 19:30:54 -04:00
commit cb18b3eb26
8 changed files with 535 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/ansible/callbacks.py
View file

@ -25,7 +25,7 @@ from ansible.color import stringc
dirname = os.path.dirname(__file__)
callbacks = utils.import_plugins(os.path.join(dirname, 'callback_plugins'))
callbacks = [ c.CallbackModule() for c in callbacks.values() ]
callbacks = [ c.CallbackModule() for c in callbacks.values() if c.__name__ != '__init__' ]
cowsay = None
if os.path.exists("/usr/bin/cowsay"):

3
lib/ansible/utils.py
View file

@ -317,6 +317,7 @@ def _gitinfo():
branch = f.readline().split('/')[-1].rstrip("\n")
f.close()
branch_path = os.path.join(repo_path, "refs", "heads", branch)
if os.path.exists(branch_path):
f = open(branch_path)
commit = f.readline()[:10]
f.close()
@ -327,6 +328,8 @@ def _gitinfo():
offset = time.altzone
result = "({0} {1}) last updated {2} (GMT {3:+04d})".format(branch, commit,
time.strftime("%Y/%m/%d %H:%M:%S", date), offset / -36)
else:
result = 'n/a'
return result
def version(prog):

22
library/git
View file

@ -45,13 +45,23 @@ def clone(repo, dest):
rc = cmd.returncode
return (rc, out, err)
def reset(dest):
def has_local_mods(dest):
os.chdir(dest)
cmd = "git status -s"
lines = os.popen(cmd).read().splitlines()
lines = filter(lambda c: re.search('^\\?\\?.*$',c) == None,lines)
return len(lines) > 0
def reset(module,dest,force):
'''
Resets the index and working tree to HEAD.
Discards any changes to tracked files in working
tree since that commit.
'''
os.chdir(dest)
if not force and has_local_mods(dest):
module.fail_json(msg="Local modifications exist in repository (force=no).")
cmd = "git reset --hard HEAD"
cmd = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = cmd.communicate()
@ -140,7 +150,8 @@ def main():
dest=dict(required=True),
repo=dict(required=True, aliases=['name']),
version=dict(default='HEAD'),
remote=dict(default='origin')
remote=dict(default='origin'),
force=dict(default='yes', choices=['yes', 'no'], aliases=['force'])
)
)
@ -148,6 +159,7 @@ def main():
repo = module.params['repo']
version = module.params['version']
remote = module.params['remote']
force = module.boolean(module.params['force'])
gitconfig = os.path.join(dest, '.git', 'config')
@ -156,14 +168,16 @@ def main():
# if there is no git configuration, do a clone operation
# else pull and switch the version
before = None
local_mods = False
if not os.path.exists(gitconfig):
(rc, out, err) = clone(repo, dest)
if rc != 0:
module.fail_json(msg=err)
else:
# else do a pull
local_mods = has_local_mods(dest)
before = get_version(dest)
(rc, out, err) = reset(dest)
(rc, out, err) = reset(module,dest,force)
if rc != 0:
module.fail_json(msg=err)
(rc, out, err) = pull(module, repo, dest, version)
@ -182,7 +196,7 @@ def main():
after = get_version(dest)
changed = False
if before != after:
if before != after or local_mods:
changed = True
module.exit_json(changed=changed, before=before, after=after)

23
library/postgresql_db
View file

@ -27,17 +27,32 @@ else:
# PostgreSQL module specific support methods.
#
def set_owner(cursor, db, owner):
query = "ALTER DATABASE %s OWNER TO %s" % (db, owner)
cursor.execute(query)
return True
def db_owned_by(cursor, db, user):
query = """SELECT * FROM pg_database JOIN pg_user ON datdba = usesysid
WHERE usename = %(user)s and datname = %(db)s"""
cursor.execute(query, {'db':db, 'user':user})
return cursor.rowcount == 1
def db_exists(cursor, db):
query = "SELECT * FROM pg_database WHERE datname=%(db)s"
cursor.execute(query, {'db': db})
return cursor.rowcount == 1
def db_delete(cursor, db):
if db_exists(cursor, db):
query = "DROP DATABASE %s" % db
cursor.execute(query)
return True
else:
return False
def db_create(cursor, db, owner, template, encoding):
if not db_exists(cursor, db):
if owner:
owner = " OWNER %s" % owner
if template:
@ -47,6 +62,10 @@ def db_create(cursor, db, owner, template, encoding):
query = "CREATE DATABASE %s%s%s%s" % (db, owner, template, encoding)
cursor.execute(query)
return True
elif owner and not db_owned_by(cursor, db, owner):
return set_owner(cursor, db, owner)
else:
return False
# ===========================================
# Module execution.
@ -100,11 +119,9 @@ def main():
module.fail_json(msg="unable to connect to database: %s" % e)
try:
if db_exists(cursor, db):
if state == "absent":
changed = db_delete(cursor, db)
else:
if state == "present":
elif state == "present":
changed = db_create(cursor, db, owner, template, encoding)
except Exception, e:
module.fail_json(msg="Database query failed: %s" % e)

223
library/postgresql_user
View file

@ -16,6 +16,8 @@
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
import re
try:
import psycopg2
except ImportError:
@ -34,35 +36,14 @@ def user_exists(cursor, user):
return cursor.rowcount > 0
def user_add(cursor, user, password, db):
def user_add(cursor, user, password):
"""Create a new user with write access to the database"""
query = "CREATE USER %(user)s with PASSWORD '%(password)s'"
cursor.execute(query % {"user": user, "password": password})
grant_privileges(cursor, user, db)
return True
def has_privileges(cursor, user, db):
"""Check if the user has create privileges on the database"""
query = "SELECT has_database_privilege(%(user)s, %(db)s, 'CREATE')"
cursor.execute(query, {'user': user, 'db': db})
return cursor.fetchone()[0]
def grant_privileges(cursor, user, db):
"""Grant all privileges on the database"""
query = "GRANT ALL PRIVILEGES ON DATABASE %(db)s TO %(user)s"
cursor.execute(query % {'user': user, 'db': db})
def revoke_privileges(cursor, user, db):
"""Revoke all privileges on the database"""
query = "REVOKE ALL PRIVILEGES ON DATABASE %(db)s FROM %(user)s"
cursor.execute(query % {'user': user, 'db': db})
def user_mod(cursor, user, password, db):
"""Update password and permissions"""
def user_chpass(cursor, user, password):
"""Change user password"""
changed = False
# Handle passwords.
@ -79,28 +60,159 @@ def user_mod(cursor, user, password, db):
if current_pass_hash != new_pass_hash:
changed = True
# Handle privileges.
# For now, we just check if the user has access to the database
if not has_privileges(cursor, user, db):
grant_privileges(cursor, user, db)
changed = True
return changed
def user_delete(cursor, user):
"""Try to remove a user. Returns True if successful otherwise False"""
cursor.execute("SAVEPOINT ansible_pgsql_user_delete")
try:
cursor.execute("DROP USER %s" % user)
except:
cursor.execute("ROLLBACK TO SAVEPOINT ansible_pgsql_user_delete")
cursor.execute("RELEASE SAVEPOINT ansible_pgsql_user_delete")
return False
cursor.execute("RELEASE SAVEPOINT ansible_pgsql_user_delete")
return True
def has_table_privilege(cursor, user, table, priv):
query = 'SELECT has_table_privilege(%s, %s, %s)'
cursor.execute(query, (user, table, priv))
return cursor.fetchone()[0]
def get_table_privileges(cursor, user, table):
if '.' in table:
schema, table = table.split('.', 1)
else:
schema = 'public'
query = '''SELECT privilege_type FROM information_schema.role_table_grants
WHERE grantee=%s AND table_name=%s AND table_schema=%s'''
cursor.execute(query, (user, table, schema))
return set([x[0] for x in cursor.fetchall()])
def grant_table_privilege(cursor, user, table, priv):
prev_priv = get_table_privileges(cursor, user, table)
query = 'GRANT %s ON TABLE %s TO %s' % (priv, table, user)
cursor.execute(query)
curr_priv = get_table_privileges(cursor, user, table)
return len(curr_priv) > len(prev_priv)
def revoke_table_privilege(cursor, user, table, priv):
prev_priv = get_table_privileges(cursor, user, table)
query = 'REVOKE %s ON TABLE %s FROM %s' % (priv, table, user)
cursor.execute(query)
curr_priv = get_table_privileges(cursor, user, table)
return len(curr_priv) < len(prev_priv)
def get_database_privileges(cursor, user, db):
priv_map = {
'C':'CREATE',
'T':'TEMPORARY',
'c':'CONNECT',
}
query = 'SELECT datacl FROM pg_database WHERE datname = %s'
cursor.execute(query, (db,))
datacl = cursor.fetchone()[0]
r = re.search('%s=(C?T?c?)/[a-z]+\,?' % user, datacl)
if r is None:
return []
o = []
for v in r.group(1):
o.append(priv_map[v])
return o
def has_database_privilege(cursor, user, db, priv):
query = 'SELECT has_database_privilege(%s, %s, %s)'
cursor.execute(query, (user, db, priv))
return cursor.fetchone()[0]
def grant_database_privilege(cursor, user, db, priv):
prev_priv = get_database_privileges(cursor, user, db)
query = 'GRANT %s ON DATABASE %s TO %s' % (priv, db, user)
cursor.execute(query)
curr_priv = get_database_privileges(cursor, user, db)
return len(curr_priv) > len(prev_priv)
def revoke_database_privilege(cursor, user, db, priv):
prev_priv = get_database_privileges(cursor, user, db)
query = 'REVOKE %s ON DATABASE %s FROM %s' % (priv, db, user)
cursor.execute(query)
curr_priv = get_database_privileges(cursor, user, db)
return len(curr_priv) < len(prev_priv)
def revoke_privileges(cursor, user, privs):
if privs is None:
return False
changed = False
for type_ in privs:
revoke_func = {
'table':revoke_table_privilege,
'database':revoke_database_privilege
}[type_]
for name, privileges in privs[type_].iteritems():
for privilege in privileges:
changed = revoke_func(cursor, user, name, privilege)\
or changed
return changed
def grant_privileges(cursor, user, privs):
if privs is None:
return False
def user_delete(cursor, user, db):
"""Delete a user, first revoking privileges"""
revoke_privileges(cursor, user, db)
cursor.execute("DROP USER %(user)s" % {'user': user})
return True
changed = False
for type_ in privs:
grant_func = {
'table':grant_table_privilege,
'database':grant_database_privilege
}[type_]
for name, privileges in privs[type_].iteritems():
for privilege in privileges:
changed = grant_func(cursor, user, name, privilege)\
or changed
return changed
def parse_privs(privs, db):
"""
Parse privilege string to determine permissions for database db.
Format:
privileges[/privileges/...]
Where:
privileges := DATABASE_PRIVILEGES[,DATABASE_PRIVILEGES,...] |
TABLE_NAME:TABLE_PRIVILEGES[,TABLE_PRIVILEGES,...]
"""
if privs is None:
return privs
o_privs = {
'database':{},
'table':{}
}
for token in privs.split('/'):
if ':' not in token:
type_ = 'database'
name = db
priv_set = set(x.strip() for x in token.split(','))
else:
type_ = 'table'
name, privileges = token.split(':', 1)
priv_set = set(x.strip() for x in privileges.split(','))
o_privs[type_][name] = priv_set
return o_privs
# ===========================================
# Module execution.
#
def main():
module = AnsibleModule(
argument_spec=dict(
@ -110,13 +222,19 @@ def main():
user=dict(required=True, aliases=['name']),
password=dict(default=None),
state=dict(default="present", choices=["absent", "present"]),
db=dict(required=True),
priv=dict(default=None),
db=dict(default=''),
fail_on_user=dict(default='yes')
)
)
user = module.params["user"]
password = module.params["password"]
state = module.params["state"]
fail_on_user = module.params["fail_on_user"] == 'yes'
db = module.params["db"]
if db == '' and module.params["priv"] is not None:
module.fail_json(msg="privileges require a database to be specified")
privs = parse_privs(module.params["priv"], db)
if not postgresqldb_found:
module.fail_json(msg="the python psycopg2 module is required")
@ -127,33 +245,44 @@ def main():
params_map = {
"login_host":"host",
"login_user":"user",
"login_password":"password"
"login_password":"password",
"db":"database"
}
kw = dict( (params_map[k], v) for (k, v) in module.params.iteritems()
if k in params_map and v != "" )
try:
db_connection = psycopg2.connect(database=db, **kw)
db_connection = psycopg2.connect(**kw)
cursor = db_connection.cursor()
except Exception, e:
module.fail_json(msg="unable to connect to database: %s" % e)
kw = dict(user=user)
changed = False
user_removed = False
if state == "present":
if user_exists(cursor, user):
changed = user_mod(cursor, user, password, db)
changed = user_chpass(cursor, user, password)
else:
if password is None:
msg = "password parameter required when adding a user"
module.fail_json(msg=msg)
changed = user_add(cursor, user, password, db)
elif state == "absent":
if user_exists(cursor, user):
changed = user_delete(cursor, user, db)
changed = user_add(cursor, user, password)
changed = grant_privileges(cursor, user, privs) or changed
else:
changed = False
# Commit the database changes
if user_exists(cursor, user):
changed = revoke_privileges(cursor, user, privs)
user_removed = user_delete(cursor, user)
changed = changed or user_removed
if fail_on_user and not user_removed:
msg = "unable to remove user"
module.fail_json(msg=msg)
kw['user_removed'] = user_removed
if changed:
db_connection.commit()
module.exit_json(changed=changed, user=user)
kw['changed'] = changed
module.exit_json(**kw)
# this is magic, see lib/ansible/module_common.py
#<<INCLUDE_ANSIBLE_MODULE_COMMON>>

152
library/subversion Normal file
View file

@ -0,0 +1,152 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2012, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# I wanted to keep this simple at first, so for now this checks out
# from the given branch of a repo at a particular SHA or
# tag. Latest is not supported, you should not be doing
# that. Contribs welcome! -- MPD
# requires subversion and grep on the client.
import re
def get_version(dest):
''' samples the version of the git repo '''
os.chdir(dest)
cmd = "svn info"
revision = filter(lambda l: re.search('Revision',l) != None,os.popen(cmd).read().splitlines())
url = filter(lambda l: re.search('^URL',l) != None,os.popen(cmd).read().splitlines())
return [revision[0],url[0]]
def checkout(repo, dest):
''' makes a new svn repo if it does not already exist '''
cmd = "svn co %s %s" % (repo, dest)
cmd = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = cmd.communicate()
rc = cmd.returncode
return (rc, out, err)
def switch(repo, dest):
''' makes a new svn repo if it does not already exist '''
cmd = "svn sw %s %s" % (repo, dest)
cmd = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = cmd.communicate()
rc = cmd.returncode
return (rc, out, err)
def has_local_mods(dest):
os.chdir(dest)
cmd = "svn status"
lines = os.popen(cmd).read().splitlines()
filtered = filter(lambda c: re.search('^\\?.*$',c) == None,lines)
return len(filtered) > 0
def reset(dest,force):
'''
Reset the repo:
force: if true, then remove any local modifications. Else, fail if there are local modifications
'''
if has_local_mods(dest):
if force:
cmd = "svn revert -R ."
cmd = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = cmd.communicate()
rc = cmd.returncode
return (rc, out, err)
else:
return (-1,"ERROR: modified files exist in the repository.","")
return (0,"","")
def update(module, dest, version):
''' update an existing svn repo '''
os.chdir(dest)
cmd = ''
if version != 'HEAD':
cmd = "svn up -r %s" % version
else:
cmd = "svn up"
cmd = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(out, err) = cmd.communicate()
rc = cmd.returncode
return (rc, out, err)
# ===========================================
def main():
module = AnsibleModule(
argument_spec = dict(
dest=dict(required=True),
repo=dict(required=True, aliases=['name']),
revision=dict(default='HEAD'),
force=dict(default='yes', choices=['yes', 'no'], aliases=['force'])
)
)
dest = module.params['dest']
repo = module.params['repo']
revision = module.params['revision']
force = module.boolean(module.params['force'])
rc, out, err, status = (0, None, None, None)
# if there is no .svn folder, do a checkout
# else update.
before = None
local_mods = False
if not os.path.exists("%s/.svn" % (dest)):
if os.path.exists(dest):
module.fail_json(msg="%s folder already exists, but its not a subversion repository." % (dest))
else:
(rc, out, err) = checkout(repo, dest)
if rc != 0:
module.fail_json(msg=err)
else:
local_mods = has_local_mods(dest)
# else do an update
before = get_version(dest)
(rc, out, err) = reset(dest,force)
if rc != 0:
module.fail_json(msg=err)
(rc, out, err) = switch(repo, dest)
if rc != 0:
module.fail_json(msg=err)
# handle errors from switch or pull
if err.find('ERROR') != -1:
module.fail_json(msg=err)
# switch to version specified regardless of whether
# we cloned or pulled
(rc, out, err) = update(module, dest, revision)
if rc != 0:
module.fail_json(msg=err)
# determine if we changed anything
after = get_version(dest)
changed = False
if before != after or local_mods:
changed = True
module.exit_json(changed=changed, before=before, after=after)
# include magic from lib/ansible/module_common.py
#<<INCLUDE_ANSIBLE_MODULE_COMMON>>
main()

59
library/wait_for Normal file
View file

@ -0,0 +1,59 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# (c) 2012, Jeroen Hoekx <jeroen@hoekx.be>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
import socket
import datetime
import time
import sys
def main():
module = AnsibleModule(
argument_spec = dict(
name=dict(required=True),
timeout=dict(default=300),
port=dict(default=22),
),
)
params = module.params
host = params['name']
timeout = int(params['timeout'])
port = int(params['port'])
end = datetime.datetime.now() + datetime.timedelta(seconds=timeout)
while datetime.datetime.now() < end:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.connect( (host, port) )
s.close()
break
except:
time.sleep(1)
else:
module.fail_json(msg="Timeout when waiting for %s"%(host))
module.exit_json(msg="%s responds on %s"%(host, port))
# this is magic, see lib/ansible/module_common.py
#<<INCLUDE_ANSIBLE_MODULE_COMMON>>
main()

81
test/TestRunner.py
View file

@ -138,6 +138,85 @@ class TestRunner(unittest.TestCase):
assert 'failed' not in result
assert result['rc'] == 0
def test_git(self):
self._run('file',['path=/tmp/gitdemo','state=absent'])
self._run('file',['path=/tmp/gd','state=absent'])
self._run('command',['git init gitdemo', 'chdir=/tmp'])
self._run('command',['touch a', 'chdir=/tmp/gitdemo'])
self._run('command',['git add *', 'chdir=/tmp/gitdemo'])
self._run('command',['git commit -m "test commit 2"', 'chdir=/tmp/gitdemo'])
self._run('command',['touch b', 'chdir=/tmp/gitdemo'])
self._run('command',['git add *', 'chdir=/tmp/gitdemo'])
self._run('command',['git commit -m "test commit 2"', 'chdir=/tmp/gitdemo'])
result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd" ])
assert result['changed']
# test the force option not set
self._run('file',['path=/tmp/gd/a','state=absent'])
result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd", "force=no" ])
assert result['failed']
# test the force option when set
result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd", "force=yes" ])
assert result['changed']
def test_subversion(self):
# TODO make an svn repo locally so as to avoid tests failing on network calls
self._run('file',['path=/tmp/meetings','state=absent'])
# hacking/test-module -m library/subversion
result = self._run('subversion', [ ])
assert result['failed']
assert "dest" in result['msg']
assert "repo" in result['msg']
# hacking/test-module -m library/subversion -a "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\""
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"" ])
assert result['failed']
assert "dest" in result['msg']
# hacking/test-module -m library/subversion -a "dest=\"/tmp/gnconf\""
result = self._run('subversion', [ "dest=\"/tmp/gnconf\"" ])
assert result['failed']
assert "repo" in result['msg']
# when /tmp/meetings doesn't exist:
# hacking/test-module -m library/subversion -a "repo=\"repo\" dest=\"/tmp/gnconf\""
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert result['changed']
# when /tmp/meetings exists, but nothing has changed.
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert not result['changed']
# when /tmp/meetings is a folder, but its not an svn repo
self._run('file',['path=/tmp/meetings','state=absent'])
self._run('file',['path=/tmp/meetings','state=directory'])
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert result['failed']
# when /tmp/meetings is a file
self._run('file',['path=/tmp/meetings','state=absent'])
self._run('command',['touch /tmp/meetings'])
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert result['failed']
# when /tmp/meetings is a folder, but its a different svn URL - should automatically switch
self._run('file',['path=/tmp/meetings','state=absent'])
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/api-errata\"","dest=\"/tmp/meetings\"" ])
assert result['changed']
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert result['changed']
assert result['after'][1] == 'URL: http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings'
# when /tmp/meetings is a folder, when its an older revision it should update
self._run('command',['svn up -r926415','chdir=/tmp/meetings'])
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert result['changed']
assert result['before'][0] == 'Revision: 926415'
assert result['after'][0] != 'Revision: 926415'
# when /tmp/meetings has dirty files in it, ignore them:
self._run('command',['touch /tmp/meetings/adirtyfile'])
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"" ])
assert not result['changed'] # no changes to the repo
# when /tmp/meetings has modified file in it, fail:
self._run('file',['path=/tmp/meetings/adirtyfile','state=absent'])
self._run('command',['cp /tmp/meetings/berlin-11-agenda /tmp/meetings/svn-vision-agenda'])
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"","force=no" ])
assert result['failed']
# when /tmp/meetings has a modified file but force is set to yes, then just override it.
result = self._run('subversion', [ "repo=\"http://svn.apache.org/repos/asf/subversion/trunk/notes/meetings\"","dest=\"/tmp/meetings\"","force=yes" ])
assert result['changed'] # no changes to the repo
def test_large_output(self):
large_path = "/usr/share/dict/words"
if not os.path.exists(large_path):
@ -198,5 +277,3 @@ class TestRunner(unittest.TestCase):
"dest=%s" % output,
])
assert result['changed'] == False