wutta/wuttamess
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: format all code with black

Browse source 
and from now on should not deviate from that...
This commit is contained in:
Lance Edgar 2025-08-31 12:52:36 -05:00
parent 2bd094b10b
commit 147d2fd871
17 changed files with 298 additions and 216 deletions
Download patch file Download diff file Expand all files Collapse all files

30
tests/test_apt.py
View file

@ -10,19 +10,23 @@ class TestDistUpgrade(TestCase):
def test_basic(self):
c = MagicMock()
with patch.object(mod, 'update') as update:
with patch.object(mod, 'upgrade') as upgrade:
mod.dist_upgrade(c, frontend='whatever')
with patch.object(mod, "update") as update:
with patch.object(mod, "upgrade") as upgrade:
mod.dist_upgrade(c, frontend="whatever")
update.assert_called_once_with(c)
upgrade.assert_called_once_with(c, dist_upgrade=True, frontend='whatever')
upgrade.assert_called_once_with(
c, dist_upgrade=True, frontend="whatever"
)
class TestInstall(TestCase):
def test_basic(self):
c = MagicMock()
mod.install(c, 'postfix')
c.run.assert_called_once_with('DEBIAN_FRONTEND=noninteractive apt-get --assume-yes install postfix')
mod.install(c, "postfix")
c.run.assert_called_once_with(
"DEBIAN_FRONTEND=noninteractive apt-get --assume-yes install postfix"
)
class TestIsInstalled(TestCase):
@ -30,14 +34,14 @@ class TestIsInstalled(TestCase):
def test_already_installed(self):
c = MagicMock()
c.run.return_value.ok = True
self.assertTrue(mod.is_installed(c, 'postfix'))
c.run.assert_called_once_with('dpkg-query -s postfix', warn=True)
self.assertTrue(mod.is_installed(c, "postfix"))
c.run.assert_called_once_with("dpkg-query -s postfix", warn=True)
def test_not_installed(self):
c = MagicMock()
c.run.return_value.ok = False
self.assertFalse(mod.is_installed(c, 'postfix'))
c.run.assert_called_once_with('dpkg-query -s postfix', warn=True)
self.assertFalse(mod.is_installed(c, "postfix"))
c.run.assert_called_once_with("dpkg-query -s postfix", warn=True)
class TestUpdate(TestCase):
@ -45,7 +49,7 @@ class TestUpdate(TestCase):
def test_basic(self):
c = MagicMock()
mod.update(c)
c.run.assert_called_once_with('apt-get update')
c.run.assert_called_once_with("apt-get update")
class TestUpgrade(TestCase):
@ -53,4 +57,6 @@ class TestUpgrade(TestCase):
def test_basic(self):
c = MagicMock()
mod.upgrade(c)
c.run.assert_called_once_with('DEBIAN_FRONTEND=noninteractive apt-get --assume-yes --option Dpkg::Options::="--force-confdef" --option Dpkg::Options::="--force-confold" upgrade')
c.run.assert_called_once_with(
'DEBIAN_FRONTEND=noninteractive apt-get --assume-yes --option Dpkg::Options::="--force-confdef" --option Dpkg::Options::="--force-confold" upgrade'
)

16
tests/test_postfix.py
View file

@ -10,7 +10,7 @@ class TestSetConfig(TestCase):
def test_basic(self):
c = MagicMock()
mod.set_config(c, 'foo', 'bar')
mod.set_config(c, "foo", "bar")
c.run.assert_called_once_with("postconf -e 'foo=bar'")
@ -18,7 +18,7 @@ class TestSetMyhostname(TestCase):
def test_basic(self):
c = MagicMock()
mod.set_myhostname(c, 'test.example.com')
mod.set_myhostname(c, "test.example.com")
c.run.assert_called_once_with("postconf -e 'myhostname=test.example.com'")
@ -26,7 +26,7 @@ class TestSetMyorigin(TestCase):
def test_basic(self):
c = MagicMock()
mod.set_myorigin(c, 'example.com')
mod.set_myorigin(c, "example.com")
c.run.assert_called_once_with("postconf -e 'myorigin=example.com'")
@ -34,15 +34,17 @@ class TestSetMydestination(TestCase):
def test_basic(self):
c = MagicMock()
mod.set_mydestination(c, 'example.com', 'test.example.com', 'localhost')
c.run.assert_called_once_with("postconf -e 'mydestination=example.com, test.example.com, localhost'")
mod.set_mydestination(c, "example.com", "test.example.com", "localhost")
c.run.assert_called_once_with(
"postconf -e 'mydestination=example.com, test.example.com, localhost'"
)
class TestSetMynetworks(TestCase):
def test_basic(self):
c = MagicMock()
mod.set_mynetworks(c, '127.0.0.0/8', '[::1]/128')
mod.set_mynetworks(c, "127.0.0.0/8", "[::1]/128")
c.run.assert_called_once_with("postconf -e 'mynetworks=127.0.0.0/8 [::1]/128'")
@ -50,5 +52,5 @@ class TestSetRelayhost(TestCase):
def test_basic(self):
c = MagicMock()
mod.set_relayhost(c, 'mail.example.com')
mod.set_relayhost(c, "mail.example.com")
c.run.assert_called_once_with("postconf -e 'relayhost=mail.example.com'")

115
tests/test_postgres.py
View file

@ -11,98 +11,115 @@ class TestSql(TestCase):
def test_basic(self):
c = MagicMock()
mod.sql(c, "select @@version")
c.sudo.assert_called_once_with('psql --tuples-only --no-align --command="select @@version" ',
user='postgres')
c.sudo.assert_called_once_with(
'psql --tuples-only --no-align --command="select @@version" ',
user="postgres",
)
class TestUserExists(TestCase):
def test_user_exists(self):
c = MagicMock()
with patch.object(mod, 'sql') as sql:
sql.return_value.stdout = 'foo'
self.assertTrue(mod.user_exists(c, 'foo'))
sql.assert_called_once_with(c, "SELECT rolname FROM pg_roles WHERE rolname = 'foo'", port=None)
with patch.object(mod, "sql") as sql:
sql.return_value.stdout = "foo"
self.assertTrue(mod.user_exists(c, "foo"))
sql.assert_called_once_with(
c, "SELECT rolname FROM pg_roles WHERE rolname = 'foo'", port=None
)
def test_user_does_not_exist(self):
c = MagicMock()
with patch.object(mod, 'sql') as sql:
sql.return_value.stdout = ''
self.assertFalse(mod.user_exists(c, 'foo'))
sql.assert_called_once_with(c, "SELECT rolname FROM pg_roles WHERE rolname = 'foo'", port=None)
with patch.object(mod, "sql") as sql:
sql.return_value.stdout = ""
self.assertFalse(mod.user_exists(c, "foo"))
sql.assert_called_once_with(
c, "SELECT rolname FROM pg_roles WHERE rolname = 'foo'", port=None
)
class TestCreateUser(TestCase):
def test_basic(self):
c = MagicMock()
with patch.object(mod, 'set_user_password') as set_user_password:
mod.create_user(c, 'foo', checkfirst=False)
c.sudo.assert_called_once_with('createuser --no-createrole --no-superuser foo',
user='postgres')
with patch.object(mod, "set_user_password") as set_user_password:
mod.create_user(c, "foo", checkfirst=False)
c.sudo.assert_called_once_with(
"createuser --no-createrole --no-superuser foo", user="postgres"
)
set_user_password.assert_not_called()
def test_user_exists(self):
c = MagicMock()
with patch.object(mod, 'user_exists') as user_exists:
with patch.object(mod, "user_exists") as user_exists:
user_exists.return_value = True
mod.create_user(c, 'foo')
user_exists.assert_called_once_with(c, 'foo', port=None)
mod.create_user(c, "foo")
user_exists.assert_called_once_with(c, "foo", port=None)
c.sudo.assert_not_called()
def test_with_password(self):
c = MagicMock()
with patch.object(mod, 'set_user_password') as set_user_password:
mod.create_user(c, 'foo', 'foopass', checkfirst=False)
c.sudo.assert_called_once_with('createuser --no-createrole --no-superuser foo',
user='postgres')
set_user_password.assert_called_once_with(c, 'foo', 'foopass', port=None)
with patch.object(mod, "set_user_password") as set_user_password:
mod.create_user(c, "foo", "foopass", checkfirst=False)
c.sudo.assert_called_once_with(
"createuser --no-createrole --no-superuser foo", user="postgres"
)
set_user_password.assert_called_once_with(c, "foo", "foopass", port=None)
class TestSetUserPassword(TestCase):
def test_basic(self):
c = MagicMock()
with patch.object(mod, 'sql') as sql:
mod.set_user_password(c, 'foo', 'foopass')
sql.assert_called_once_with(c, "ALTER USER \\\"foo\\\" PASSWORD 'foopass';",
port=None, hide=True, echo=False)
with patch.object(mod, "sql") as sql:
mod.set_user_password(c, "foo", "foopass")
sql.assert_called_once_with(
c,
"ALTER USER \\\"foo\\\" PASSWORD 'foopass';",
port=None,
hide=True,
echo=False,
)
class TestDbExists(TestCase):
def test_db_exists(self):
c = MagicMock()
with patch.object(mod, 'sql') as sql:
sql.return_value.stdout = 'foo'
self.assertTrue(mod.db_exists(c, 'foo'))
sql.assert_called_once_with(c, "SELECT datname FROM pg_database WHERE datname = 'foo'", port=None)
with patch.object(mod, "sql") as sql:
sql.return_value.stdout = "foo"
self.assertTrue(mod.db_exists(c, "foo"))
sql.assert_called_once_with(
c, "SELECT datname FROM pg_database WHERE datname = 'foo'", port=None
)
def test_db_does_not_exist(self):
c = MagicMock()
with patch.object(mod, 'sql') as sql:
sql.return_value.stdout = ''
self.assertFalse(mod.db_exists(c, 'foo'))
sql.assert_called_once_with(c, "SELECT datname FROM pg_database WHERE datname = 'foo'", port=None)
with patch.object(mod, "sql") as sql:
sql.return_value.stdout = ""
self.assertFalse(mod.db_exists(c, "foo"))
sql.assert_called_once_with(
c, "SELECT datname FROM pg_database WHERE datname = 'foo'", port=None
)
class TestCreateDb(TestCase):
def test_basic(self):
c = MagicMock()
mod.create_db(c, 'foo', checkfirst=False)
c.sudo.assert_called_once_with('createdb foo', user='postgres')
mod.create_db(c, "foo", checkfirst=False)
c.sudo.assert_called_once_with("createdb foo", user="postgres")
def test_db_exists(self):
c = MagicMock()
with patch.object(mod, 'db_exists') as db_exists:
with patch.object(mod, "db_exists") as db_exists:
db_exists.return_value = True
mod.create_db(c, 'foo')
db_exists.assert_called_once_with(c, 'foo', port=None)
mod.create_db(c, "foo")
db_exists.assert_called_once_with(c, "foo", port=None)
c.sudo.assert_not_called()
@ -110,17 +127,17 @@ class TestDropDb(TestCase):
def test_basic(self):
c = MagicMock()
mod.drop_db(c, 'foo', checkfirst=False)
c.sudo.assert_called_once_with('dropdb foo', user='postgres')
mod.drop_db(c, "foo", checkfirst=False)
c.sudo.assert_called_once_with("dropdb foo", user="postgres")
def test_db_does_not_exist(self):
c = MagicMock()
with patch.object(mod, 'db_exists') as db_exists:
with patch.object(mod, "db_exists") as db_exists:
db_exists.return_value = False
mod.drop_db(c, 'foo')
db_exists.assert_called_once_with(c, 'foo')
mod.drop_db(c, "foo")
db_exists.assert_called_once_with(c, "foo")
c.sudo.assert_not_called()
@ -128,7 +145,9 @@ class TestDumpDb(TestCase):
def test_basic(self):
c = MagicMock()
result = mod.dump_db(c, 'foo')
self.assertEqual(result, 'foo.sql.gz')
c.sudo.assert_called_once_with('pg_dump foo | gzip -c > /tmp/foo.sql.gz', user='postgres')
c.run.assert_called_with('rm /tmp/foo.sql.gz')
result = mod.dump_db(c, "foo")
self.assertEqual(result, "foo.sql.gz")
c.sudo.assert_called_once_with(
"pg_dump foo | gzip -c > /tmp/foo.sql.gz", user="postgres"
)
c.run.assert_called_with("rm /tmp/foo.sql.gz")

33
tests/test_ssh.py
View file

@ -13,8 +13,8 @@ class TestCacheHostKey(TestCase):
# assume the first command runs okay
c.run.return_value.failed = False
mod.cache_host_key(c, 'example.com')
c.run.assert_called_once_with('ssh example.com whoami', warn=True)
mod.cache_host_key(c, "example.com")
c.run.assert_called_once_with("ssh example.com whoami", warn=True)
def test_root_commands_not_allowed(self):
c = MagicMock()
@ -22,25 +22,27 @@ class TestCacheHostKey(TestCase):
# assume the first command fails b/c "disallowed"
c.run.return_value.failed = True
c.run.return_value.stderr = "Disallowed command"
mod.cache_host_key(c, 'example.com')
c.run.assert_called_once_with('ssh example.com whoami', warn=True)
mod.cache_host_key(c, "example.com")
c.run.assert_called_once_with("ssh example.com whoami", warn=True)
def test_root_cache_key(self):
c = MagicMock()
# first command fails; second command caches host key
c.run.return_value.failed = True
mod.cache_host_key(c, 'example.com')
c.run.assert_has_calls([call('ssh example.com whoami', warn=True)])
c.run.assert_called_with('ssh -o StrictHostKeyChecking=no example.com whoami', warn=True)
mod.cache_host_key(c, "example.com")
c.run.assert_has_calls([call("ssh example.com whoami", warn=True)])
c.run.assert_called_with(
"ssh -o StrictHostKeyChecking=no example.com whoami", warn=True
)
def test_user_already_cached(self):
c = MagicMock()
# assume the first command runs okay
c.sudo.return_value.failed = False
mod.cache_host_key(c, 'example.com', user='foo')
c.sudo.assert_called_once_with('ssh example.com whoami', user='foo', warn=True)
mod.cache_host_key(c, "example.com", user="foo")
c.sudo.assert_called_once_with("ssh example.com whoami", user="foo", warn=True)
def test_user_commands_not_allowed(self):
c = MagicMock()
@ -48,15 +50,16 @@ class TestCacheHostKey(TestCase):
# assume the first command fails b/c "disallowed"
c.sudo.return_value.failed = True
c.sudo.return_value.stderr = "Disallowed command"
mod.cache_host_key(c, 'example.com', user='foo')
c.sudo.assert_called_once_with('ssh example.com whoami', user='foo', warn=True)
mod.cache_host_key(c, "example.com", user="foo")
c.sudo.assert_called_once_with("ssh example.com whoami", user="foo", warn=True)
def test_user_cache_key(self):
c = MagicMock()
# first command fails; second command caches host key
c.sudo.return_value.failed = True
mod.cache_host_key(c, 'example.com', user='foo')
c.sudo.assert_has_calls([call('ssh example.com whoami', user='foo', warn=True)])
c.sudo.assert_called_with('ssh -o StrictHostKeyChecking=no example.com whoami',
user='foo', warn=True)
mod.cache_host_key(c, "example.com", user="foo")
c.sudo.assert_has_calls([call("ssh example.com whoami", user="foo", warn=True)])
c.sudo.assert_called_with(
"ssh -o StrictHostKeyChecking=no example.com whoami", user="foo", warn=True
)

40
tests/test_sync.py
View file

@ -12,26 +12,26 @@ from wuttamess import sync as mod
class TestMakeRoot(TestCase):
def test_basic(self):
root = mod.make_root('files')
root = mod.make_root("files")
self.assertIsInstance(root, SyncedRoot)
self.assertEqual(root.src, Path('files'))
self.assertEqual(root.dest, Path('/'))
self.assertEqual(root.src, Path("files"))
self.assertEqual(root.dest, Path("/"))
class TestMakeSelector(TestCase):
def test_basic(self):
selector = mod.make_selector('etc/postfix')
selector = mod.make_selector("etc/postfix")
self.assertIsInstance(selector, ItemSelector)
self.assertEqual(selector.subpath, Path('etc/postfix'))
self.assertEqual(selector.subpath, Path("etc/postfix"))
class TestIsync(TestCase):
def test_basic(self):
c = MagicMock()
root = mod.make_root('files')
with patch.object(mod, 'fabsync') as fabsync:
root = mod.make_root("files")
with patch.object(mod, "fabsync") as fabsync:
fabsync.ItemSelector = ItemSelector
# nothing to sync
@ -42,7 +42,7 @@ class TestIsync(TestCase):
# sync one file
fabsync.isync.reset_mock()
result = MagicMock(path='/foo', modified=True)
result = MagicMock(path="/foo", modified=True)
fabsync.isync.return_value = [result]
results = list(mod.isync(c, root))
self.assertEqual(results, [result])
@ -50,34 +50,38 @@ class TestIsync(TestCase):
# sync with selector (subpath)
fabsync.isync.reset_mock()
result = MagicMock(path='/foo', modified=True)
result = MagicMock(path="/foo", modified=True)
fabsync.isync.return_value = [result]
results = list(mod.isync(c, root, 'foo'))
results = list(mod.isync(c, root, "foo"))
self.assertEqual(results, [result])
fabsync.isync.assert_called_once_with(c, root, selector=fabsync.ItemSelector.new('foo'))
fabsync.isync.assert_called_once_with(
c, root, selector=fabsync.ItemSelector.new("foo")
)
# sync with selector (subpath + tags)
fabsync.isync.reset_mock()
result = MagicMock(path='/foo', modified=True)
result = MagicMock(path="/foo", modified=True)
fabsync.isync.return_value = [result]
results = list(mod.isync(c, root, 'foo', tags={'bar'}))
results = list(mod.isync(c, root, "foo", tags={"bar"}))
self.assertEqual(results, [result])
fabsync.isync.assert_called_once_with(c, root, selector=fabsync.ItemSelector.new('foo', tags={'bar'}))
fabsync.isync.assert_called_once_with(
c, root, selector=fabsync.ItemSelector.new("foo", tags={"bar"})
)
class TestCheckIsync(TestCase):
def test_basic(self):
c = MagicMock()
root = mod.make_root('files')
with patch.object(mod, 'isync') as isync:
root = mod.make_root("files")
with patch.object(mod, "isync") as isync:
# file(s) modified
result = MagicMock(path='/foo', modified=True)
result = MagicMock(path="/foo", modified=True)
isync.return_value = [result]
self.assertTrue(mod.check_isync(c, root))
# not modified
result = MagicMock(path='/foo', modified=False)
result = MagicMock(path="/foo", modified=False)
isync.return_value = [result]
self.assertFalse(mod.check_isync(c, root))

52
tests/test_util.py
View file

@ -11,17 +11,17 @@ class TestExists(TestCase):
def test_basic(self):
c = MagicMock()
mod.exists(c, '/foo')
c.run.assert_called_once_with('test -e /foo', warn=True)
mod.exists(c, "/foo")
c.run.assert_called_once_with("test -e /foo", warn=True)
class TestHomePath(TestCase):
def test_basic(self):
c = MagicMock()
c.run.return_value.stdout = '/home/foo'
path = mod.get_home_path(c, user='foo')
self.assertEqual(path, '/home/foo')
c.run.return_value.stdout = "/home/foo"
path = mod.get_home_path(c, user="foo")
self.assertEqual(path, "/home/foo")
class TestIsSymlink(TestCase):
@ -29,13 +29,13 @@ class TestIsSymlink(TestCase):
def test_yes(self):
c = MagicMock()
c.run.return_value.failed = False
self.assertTrue(mod.is_symlink(c, '/foo'))
self.assertTrue(mod.is_symlink(c, "/foo"))
c.run.assert_called_once_with('test -L "$(echo /foo)"', warn=True)
def test_no(self):
c = MagicMock()
c.run.return_value.failed = True
self.assertFalse(mod.is_symlink(c, '/foo'))
self.assertFalse(mod.is_symlink(c, "/foo"))
c.run.assert_called_once_with('test -L "$(echo /foo)"', warn=True)
@ -43,31 +43,39 @@ class TestMakoRenderer(TestCase):
def test_basic(self):
c = MagicMock()
renderer = mod.mako_renderer(c, env={'machine_is_live': True})
renderer = mod.mako_renderer(c, env={"machine_is_live": True})
here = os.path.dirname(__file__)
path = os.path.join(here, 'files', 'bar', 'baz')
path = os.path.join(here, "files", "bar", "baz")
rendered = renderer(path, vars={})
self.assertEqual(rendered, 'machine_is_live = True')
self.assertEqual(rendered, "machine_is_live = True")
class TestSetTimezone(TestCase):
def test_symlink(self):
c = MagicMock()
with patch.object(mod, 'is_symlink') as is_symlink:
with patch.object(mod, "is_symlink") as is_symlink:
is_symlink.return_value = True
mod.set_timezone(c, 'America/Chicago')
c.run.assert_has_calls([
call("bash -c 'echo America/Chicago > /etc/timezone'"),
])
c.run.assert_called_with('ln -sf /usr/share/zoneinfo/America/Chicago /etc/localtime')
mod.set_timezone(c, "America/Chicago")
c.run.assert_has_calls(
[
call("bash -c 'echo America/Chicago > /etc/timezone'"),
]
)
c.run.assert_called_with(
"ln -sf /usr/share/zoneinfo/America/Chicago /etc/localtime"
)
def test_not_symlink(self):
c = MagicMock()
with patch.object(mod, 'is_symlink') as is_symlink:
with patch.object(mod, "is_symlink") as is_symlink:
is_symlink.return_value = False
mod.set_timezone(c, 'America/Chicago')
c.run.assert_has_calls([
call("bash -c 'echo America/Chicago > /etc/timezone'"),
])
c.run.assert_called_with('cp /usr/share/zoneinfo/America/Chicago /etc/localtime')
mod.set_timezone(c, "America/Chicago")
c.run.assert_has_calls(
[
call("bash -c 'echo America/Chicago > /etc/timezone'"),
]
)
c.run.assert_called_with(
"cp /usr/share/zoneinfo/America/Chicago /etc/localtime"
)

38
tests/test_wutta.py
View file

@ -12,15 +12,29 @@ class TestPurgeEmailSettings(TestCase):
c = MagicMock()
sql = MagicMock()
postgres = MagicMock(sql=sql)
with patch.object(mod, 'postgres', new=postgres):
mod.purge_email_settings(c, 'testy', appname='wuttatest')
sql.assert_has_calls([
call(c, "delete from setting where name like 'wuttatest.email.%.sender';",
database='testy'),
call(c, "delete from setting where name like 'wuttatest.email.%.to';",
database='testy'),
call(c, "delete from setting where name like 'wuttatest.email.%.cc';",
database='testy'),
call(c, "delete from setting where name like 'wuttatest.email.%.bcc';",
database='testy'),
])
with patch.object(mod, "postgres", new=postgres):
mod.purge_email_settings(c, "testy", appname="wuttatest")
sql.assert_has_calls(
[
call(
c,
"delete from setting where name like 'wuttatest.email.%.sender';",
database="testy",
),
call(
c,
"delete from setting where name like 'wuttatest.email.%.to';",
database="testy",
),
call(
c,
"delete from setting where name like 'wuttatest.email.%.cc';",
database="testy",
),
call(
c,
"delete from setting where name like 'wuttatest.email.%.bcc';",
database="testy",
),
]
)