sam/shlax
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'mostadvanced' into 'master'

Browse Source 
Complete core rewrite, with documentation

See merge request oss/shlax!1
This commit is contained in:
jpic ∞ 2021-04-24 13:07:37 +00:00
commit 3b6650efdb
46 changed files with 1715 additions and 1145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
.gitlab-ci.yml
View File

@ -1,15 +1,36 @@
build:
cache:
key: cache
paths: [.cache]
image: yourlabs/shlax
script: pip install -U --user -e . && CACHE_DIR=$(pwd)/.cache ./shlaxfile.py -d
shlax build push
paths: [.cache, /var/lib/containers/]
image: yourlabs/buildah
script:
- pip3 install -U --user .[cli]
- CACHE_DIR=$(pwd)/.cache python3 ./shlaxfile.py build push=docker://docker.io/yourlabs/shlax:$CI_COMMIT_SHORT_SHA
stage: build
build-itself:
cache:
key: cache
paths: [.cache, /var/lib/containers/]
image: yourlabs/shlax:$CI_COMMIT_SHORT_SHA
script: python3 ./shlaxfile.py build push=docker://docker.io/yourlabs/shlax:$CI_COMMIT_REF
stage: test
test-exitcode:
image: yourlabs/shlax:$CI_COMMIT_SHORT_SHA
script:
- tests/shlaxfail.py build || [ $? -eq 1 ]
- tests/shlaxsuccess.py build
test:
image: yourlabs/python
stage: build
script:
- pip install -U --user -e .[test]
- py.test -sv tests
pypi:
image: yourlabs/python
only: [tags]
script: pypi-release
stage: deploy
test: {image: yourlabs/python, script: 'pip install -U --user -e .[test] && py.test
-svv tests', stage: build}

282
README.md Normal file
View File

@ -0,0 +1,282 @@
# Shlax: Pythonic automation tool
Shlax is a Python framework for system automation, initially with the purpose
of replacing docker, docker-compose and ansible with a single tool with the
purpose of code-reuse made possible by target abstraction.
## Development status: Design state
I got the thing to work with an ugly PoC that I basically brute-forced, I'm
currently rewriting the codebase with a proper design.
The stories are in development in this order:
- replacing docker build, that's in the state of polishing
- replacing docker-compose, not in use but the PoC works so far
- replacing ansible, also working in working PoC state, the shlax command line
demonstrates
This project is supposed to unblock me from adding the CI feature to the
Sentry/GitLab/Portainer implementation I'm doing in pure python on top of
Django, CRUDLFA+ and Ryzom (isomorphic components in Python to replace
templates). So, as you can see, I'm really deep in it with a strong
determination.
Shlax builds its container itself, so check the shlaxfile.py of this repository
to see what it currently looks like, and check the build job of the CI pipeline
to see the output.
# Design
The pattern resolves around two moving parts: Actions and Targets.
## Action
An action is a function that takes a target argument, it may execute nested
actions by passing over the target argument which collects the results.
Example:
```python
async def hello_world(target):
"""Bunch of silly commands to demonstrate action programming."""
await target.mkdir('foo')
python = await target.which('python3', 'python')
await target.exec(f'{python} --version > foo/test')
version = target.exec('cat foo/test').output
print('version')
```
### Recursion
An action may call other actions recursively. There are two ways:
```python
async def something(target):
# just run the other action code
hello_world(target)
# or delegate the call to target
target(hello_world)
```
In the first case, the resulting count of ran actions will remain 1:
"something" action.
In the second case, the resulting count of ran actions will be 2: "something"
and "hello_world".
### Callable classes
Actually in practice, Actions are basic callable Python classes, here's a basic
example to run a command:
```python
class Run:
def __init__(self, cmd):
self.cmd = cmd
async def __call__(self, target):
return await target.exec(self.cmd)
```
This allows to create callable objects which may be called just like functions
and as such be appropriate actions, instead of:
```python
async def one(target):
target.exec('one')
async def two(target):
target.exec('two')
```
You can do:
```python
one = Run('one')
two = Run('two')
```
### Parallel execution
Actions may be executed in parallel with an action named ... Parallel. This
defines an action that will execute three actions in parallel:
```python
action = Parallel(
hello_world,
something,
Run('echo hi'),
)
```
In this case, all actions must succeed for the parallel action to be considered
a success.
### Methods
An action may also be a method, as long as it just takes a target argument, for
example:
```python
class Thing:
def start(self, target):
"""Starts thing"""
def stop(self, target):
"""Stops thing"""
action = Thing().start
```
### Cleaning
If an action defines a `clean` method, it will always be called wether or not
the action succeeded. Example:
```python
class Thing:
def __call__(self, target):
"""Do some thing"""
def clean(self, target):
"""Clean-up target after __call__"""
```
### Colorful actions
If an action defines a `colorize` method, it will be called with the colorset
as argument for every output, this allows to code custom output rendering.
## Target
A Target is mainly an object providing an abstraction layer over the system we
want to automate with actions. It defines functions to execute a command, mount
a directory, copy a file, manage environment variables and so on.
### Pre-configuration
A Target can be pre-configured with a list of Actions in which case calling the
target without argument will execute its Actions until one fails by raising an
Exception:
```python
say_hello = Localhost(
hello_world,
Run('echo hi'),
)
await say_hello()
```
### Results
Every time a target execute an action, it will set the "status" attribute on it
to "success" or "failure", and add it to the "results" attribute:
```python
say_hello = Localhost(Run('echo hi'))
await say_hello()
say_hello.results # contains the action with status="success"
```
## Targets as Actions: the nesting story
We've seen that any callable taking a target argument is good to be considered
an action, and that targets are callables.
To make a Target runnable like any action, all we had to do is add the target
keyword argument to `Target.__call__`.
But `target()` fills `self.results`, so nested action results would not
propagate to the parent target.
That's why if Target receives a non-None target argument, it will has to set
`self.parent` with it.
This allows nested targets to traverse parents and get to the root Target
with `target.caller`, where it can then attach results to.
This opens the nice side effect that a target implementation may call the
parent target if any, you could write a Docker target as such:
```python
class Docker(Target):
def __init__(self, *actions, name):
self.name = name
super().__init__(*actions)
async def exec(self, *args):
return await self.parent.exec(*['docker', 'exec', self.name] + args)
```
This also means that you always need a parent with an exec implementation,
there are two:
- Localhost, executes on localhost
- Stub, for testing
The result of that design is that the following use cases are available:
```python
# This action installs my favorite package on any distro
action = Packages('python3')
# Run it right here: apt install python3
Localhost()(action)
# Or remotely: ssh yourhost apt install python3
Ssh(host='yourhost')(action)
# Let's make a container build receipe with that action
build = Buildah(package)
# Run it locally: buildah exec apt install python3
Localhost()(build)
# Or on a server: ssh yourhost build exec apt install python3
Ssh(host='yourhost')(build)
# Or on a server behingh a bastion:
# ssh yourbastion ssh yourhost build exec apt install python3
Localhost()(Ssh(host='bastion')(Ssh(host='yourhost')(build))
# That's going to do the same
Localhost(Ssh(
Ssh(
build,
host='yourhost'
),
host='bastion'
))()
```
## CLI
You can execute Shlax actions directly on the command line with the `shlax` CLI
command.
For your own Shlaxfiles, you can build your CLI with your favorite CLI
framework. If you decide to use `cli2`, then Shlax provides a thin layer on top
of it: Group and Command objects made for Shlax objects.
For example:
```python
yourcontainer = Container(
build=Buildah(
User('app', '/app', 1000),
Packages('python', 'unzip', 'findutils'),
Copy('setup.py', 'yourdir', '/app'),
base='archlinux',
commit='yourimage',
),
)
if __name__ == '__main__':
print(Group(doc=__doc__).load(yourcontainer).entry_point())
```
The above will execute a cli2 command with each method of yourcontainer as a
sub-command.

7
setup.py
View File

@ -5,10 +5,9 @@ setup(
name='shlax',
versioning='dev',
setup_requires='setupmeta',
install_requires=['cli2'],
extras_require=dict(
full=[
'pyyaml',
cli=[
'cli2>=2.3.0',
],
test=[
'pytest',
@ -25,7 +24,7 @@ setup(
python_requires='>=3',
entry_points={
'console_scripts': [
'shlax = shlax.cli:cli',
'shlax = shlax.cli:cli.entry_point',
],
},
)

7
shlax/__init__.py
View File

@ -1,7 +0,0 @@
from .actions import *
from .image import Image
from .strategies import *
from .output import Output
from .proc import Proc
from .targets import *
from .shlaxfile import Shlaxfile

6
shlax/actions/__init__.py
View File

@ -1,6 +0,0 @@
from .copy import Copy
from .packages import Packages # noqa
from .base import Action # noqa
from .run import Run # noqa
from .pip import Pip
from .service import Service

211
shlax/actions/base.py
View File

@ -1,211 +1,2 @@
import functools
import inspect
import importlib
import sys
from ..output import Output
from ..exceptions import WrongResult
class Action:
parent = None
contextualize = []
regexps = {
r'([\w]+):': '{cyan}\\1{gray}:{reset}',
r'(^|\n)( *)\- ': '\\1\\2{red}-{reset} ',
}
options = dict(
debug=dict(
alias='d',
default='visit',
help='''
Display debug output. Supports values (combinable): cmd,out,visit
'''.strip(),
immediate=True,
),
)
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
@property
def context(self):
if not self.parent:
if '_context' not in self.__dict__:
self._context = dict()
return self._context
else:
return self.parent.context
def actions_filter(self, results, f=None, **filters):
if f:
def ff(a):
try:
return f(a)
except:
return False
results = [*filter(ff, results)]
for k, v in filters.items():
if k == 'type':
results = [*filter(
lambda s: type(s).__name__.lower() == str(v).lower(),
results
)]
else:
results = [*filter(
lambda s: getattr(s, k, None) == v,
results
)]
return results
def sibblings(self, f=None, **filters):
if not self.parent:
return []
return self.actions_filter(
[a for a in self.parent.actions if a is not self],
f, **filters
)
def parents(self, f=None, **filters):
if self.parent:
return self.actions_filter(
[self.parent] + self.parent.parents(),
f, **filters
)
return []
def children(self, f=None, **filters):
children = []
def add(parent):
if parent != self:
children.append(parent)
if 'actions' not in parent.__dict__:
return
for action in parent.actions:
add(action)
add(self)
return self.actions_filter(children, f, **filters)
def __getattr__(self, name):
for a in self.parents() + self.sibblings() + self.children():
if name in a.contextualize:
return getattr(a, name)
raise AttributeError(f'{type(self).__name__} has no {name}')
async def call(self, *args, **kwargs):
print(f'{self}.call(*args, **kwargs) not implemented')
sys.exit(1)
def output_factory(self, *args, **kwargs):
kwargs.setdefault('regexps', self.regexps)
return Output(**kwargs)
async def __call__(self, *args, **kwargs):
self.call_args = args
self.call_kwargs = kwargs
self.output = self.output_factory(*args, **kwargs)
self.output_start()
self.status = 'running'
try:
result = await self.call(*args, **kwargs)
except Exception as e:
self.output_fail(e)
self.status = 'fail'
proc = getattr(e, 'proc', None)
if proc:
result = proc.rc
else:
raise
else:
self.output_success()
if self.status == 'running':
self.status = 'success'
finally:
clean = getattr(self, 'clean', None)
if clean:
self.output.clean(self)
await clean(*args, **kwargs)
return result
def output_start(self):
if self.kwargs.get('quiet', False):
return
self.output.start(self)
def output_fail(self, exception=None):
if self.kwargs.get('quiet', False):
return
self.output.fail(self, exception)
def output_success(self):
if self.kwargs.get('quiet', False):
return
self.output.success(self)
def __repr__(self):
return ' '.join([type(self).__name__] + list(self.args) + [
f'{k}={v}'
for k, v in self.kwargs.items()
])
def colorized(self):
return ' '.join([
self.output.colors['pink1']
+ type(self).__name__
+ self.output.colors['yellow']
] + list(self.args) + [
f'{self.output.colors["blue"]}{k}{self.output.colors["gray"]}={self.output.colors["green2"]}{v}'
for k, v in self.kwargs_output().items()
] + [self.output.colors['reset']])
def callable(self):
from ..targets import Localhost
async def cb(*a, **k):
from shlax.cli import cli
script = Localhost(self, quiet=True)
result = await script(*a, **k)
success = functools.reduce(
lambda a, b: a + b,
[1 for c in script.children() if c.status == 'success'] or [0])
if success:
script.output.success(f'{success} PASS')
failures = functools.reduce(
lambda a, b: a + b,
[1 for c in script.children() if c.status == 'fail'] or [0])
if failures:
script.output.fail(f'{failures} FAIL')
cli.exit_code = failures
return result
return cb
def kwargs_output(self):
return self.kwargs
def action(self, action, *args, **kwargs):
if isinstance(action, str):
import cli2
a = cli2.Callable.factory(action).target
if not a:
a = cli2.Callable.factory(
'.'.join(['shlax', action])
).target
if a:
action = a
p = action(*args, **kwargs)
for parent in self.parents():
if hasattr(parent, 'actions'):
break
p.parent = parent
if 'actions' not in self.__dict__:
# "mutate" to Strategy
from ..strategies.script import Actions
self.actions = Actions(self, [p])
return p
pass

65
shlax/actions/copy.py
View File

@ -1,6 +1,63 @@
from .base import Action
import asyncio
import binascii
import glob
import os
class Copy(Action):
async def call(self, *args, **kwargs):
await self.copy(*self.args)
class Copy:
def __init__(self, *args):
self.dst = args[-1]
self.src = []
for src in args[:-1]:
if '*' in src:
self.src += glob.glob(src)
else:
self.src.append(src)
def listfiles(self):
if getattr(self, '_listfiles', None):
return self._listfiles
result = []
for src in self.src:
if os.path.isfile(src):
result.append(src)
continue
for root, dirs, files in os.walk(src):
if '__pycache__' in root:
continue
result += [
os.path.join(root, f)
for f in files
if not f.endswith('.pyc')
]
self._listfiles = result
return result
async def __call__(self, target):
await target.mkdir(self.dst)
for path in self.listfiles():
if os.path.isdir(path):
await target.mkdir(os.path.join(self.dst, path))
elif '/' in path:
dirname = os.path.join(
self.dst,
'/'.join(path.split('/')[:-1])
)
await target.mkdir(dirname)
await target.copy(path, dirname)
else:
await target.copy(path, self.dst)
def __str__(self):
return f'Copy({", ".join(self.src)}, {self.dst})'
async def cachekey(self):
async def chksum(path):
with open(path, 'rb') as f:
return (path, str(binascii.crc32(f.read())))
results = await asyncio.gather(*[chksum(f) for f in self.listfiles()])
return {path: chks for path, chks in results}

120
shlax/actions/packages.py
View File

@ -7,19 +7,16 @@ import os
import subprocess
from textwrap import dedent
from .base import Action
class Packages(Action):
class Packages:
"""
The Packages visitor wraps around the container's package manager.
Package manager abstract layer with caching.
It's a central piece of the build process, and does iterate over other
container visitors in order to pick up packages. For example, the Pip
visitor will declare ``self.packages = dict(apt=['python3-pip'])``, and the
Packages visitor will pick it up.
"""
contextualize = ['mgr']
regexps = {
#r'Installing ([\w\d-]+)': '{cyan}\\1',
r'Installing': '{cyan}lol',
@ -40,11 +37,13 @@ class Packages(Action):
update='pacman -Sy',
upgrade='pacman -Su --noconfirm',
install='pacman -S --noconfirm',
lastupdate='stat -c %Y /var/lib/pacman/sync/core.db',
),
dnf=dict(
update='dnf makecache --assumeyes',
upgrade='dnf upgrade --best --assumeyes --skip-broken', # noqa
install='dnf install --setopt=install_weak_deps=False --best --assumeyes', # noqa
lastupdate='stat -c %Y /var/cache/dnf/* | head -n1',
),
yum=dict(
update='yum update',
@ -55,63 +54,61 @@ class Packages(Action):
installed = []
def __init__(self, *packages, **kwargs):
def __init__(self, *packages, upgrade=False):
self.packages = []
self.upgrade = upgrade
for package in packages:
line = dedent(package).strip().replace('\n', ' ')
self.packages += line.split(' ')
super().__init__(*packages, **kwargs)
@property
def cache_root(self):
async def cache_setup(self, target):
if 'CACHE_DIR' in os.environ:
return os.path.join(os.getenv('CACHE_DIR'))
self.cache_root = os.path.join(os.getenv('CACHE_DIR'))
else:
return os.path.join(os.getenv('HOME'), '.cache')
self.cache_root = os.path.join(await target.parent.getenv('HOME'), '.cache')
async def update(self):
# run pkgmgr_setup functions ie. apk_setup
cachedir = await getattr(self, self.mgr + '_setup')()
await getattr(self, self.mgr + '_setup')(target)
async def update(self, target):
# lastupdate = await target.exec(self.cmds['lastupdate'], raises=False)
# lastupdate = int(lastupdate.out) if lastupdate.rc == 0 else None
lastupdate = None
if os.path.exists(cachedir + '/lastupdate'):
with open(cachedir + '/lastupdate', 'r') as f:
try:
lastupdate = int(f.read().strip())
except:
pass
if not os.path.exists(cachedir):
os.makedirs(cachedir)
now = int(datetime.now().strftime('%s'))
# cache for a week
if not lastupdate or now - lastupdate > 604800:
await target.rexec(self.cmds['update'])
return
# disabling with the above return call until needed again
# might have to rewrite this to not have our own lockfile
# or find a better place on the filesystem
# also make sure the lockfile is actually needed when running on
# targets that don't have isguest=True
if not lastupdate or now - lastupdate > 604800:
# crude lockfile implementation, should work against *most*
# race-conditions ...
lockfile = cachedir + '/update.lock'
if not os.path.exists(lockfile):
with open(lockfile, 'w+') as f:
f.write(str(os.getpid()))
if not await target.parent.exists(lockfile):
await target.parent.write(lockfile, str(os.getpid()))
try:
await self.rexec(self.cmds['update'])
await target.rexec(self.cmds['update'])
finally:
os.unlink(lockfile)
await target.parent.rm(lockfile)
with open(cachedir + '/lastupdate', 'w+') as f:
f.write(str(now))
await target.parent.write(cachedir + '/lastupdate', str(now))
else:
while os.path.exists(lockfile):
print(f'{self.container.name} | Waiting for update ...')
while await target.parent.exists(lockfile):
print(f'{self.target} | Waiting for {lockfile} ...')
await asyncio.sleep(1)
async def call(self, *args, **kwargs):
cached = getattr(self, '_pagkages_mgr', None)
async def __call__(self, target):
cached = getattr(target, 'pkgmgr', None)
if cached:
self.mgr = cached
else:
mgr = await self.which(*self.mgrs.keys())
mgr = await target.which(*self.mgrs.keys())
if mgr:
self.mgr = mgr[0].split('/')[-1]
@ -119,11 +116,15 @@ class Packages(Action):
raise Exception('Packages does not yet support this distro')
self.cmds = self.mgrs[self.mgr]
if not getattr(self, '_packages_upgraded', None):
await self.update()
if self.kwargs.get('upgrade', True):
await self.rexec(self.cmds['upgrade'])
self._packages_upgraded = True
if target.isguest:
# we're going to mount
await self.cache_setup(target)
await self.update(target)
if self.upgrade:
await target.rexec(self.cmds['upgrade'])
packages = []
for package in self.packages:
@ -136,35 +137,40 @@ class Packages(Action):
else:
packages.append(package)
await self.rexec(*self.cmds['install'].split(' ') + packages)
await target.rexec(*self.cmds['install'].split(' ') + packages)
async def apk_setup(self):
async def apk_setup(self, target):
cachedir = os.path.join(self.cache_root, self.mgr)
await self.mount(cachedir, '/var/cache/apk')
await target.mount(cachedir, '/var/cache/apk')
# special step to enable apk cache
await self.rexec('ln -sf /var/cache/apk /etc/apk/cache')
await target.rexec('ln -sf /var/cache/apk /etc/apk/cache')
return cachedir
async def dnf_setup(self):
async def dnf_setup(self, target):
cachedir = os.path.join(self.cache_root, self.mgr)
await self.mount(cachedir, f'/var/cache/{self.mgr}')
await self.rexec('echo keepcache=True >> /etc/dnf/dnf.conf')
await target.mount(cachedir, f'/var/cache/{self.mgr}')
await target.rexec('echo keepcache=True >> /etc/dnf/dnf.conf')
return cachedir
async def apt_setup(self):
codename = (await self.rexec(
f'source {self.mnt}/etc/os-release; echo $VERSION_CODENAME'
async def apt_setup(self, target):
codename = (await target.rexec(
f'source /etc/os-release; echo $VERSION_CODENAME'
)).out
cachedir = os.path.join(self.cache_root, self.mgr, codename)
await self.rexec('rm /etc/apt/apt.conf.d/docker-clean')
cache_archives = os.path.join(cachedir, 'archives')
await self.mount(cache_archives, f'/var/cache/apt/archives')
await target.mount(cache_archives, f'/var/cache/apt/archives')
cache_lists = os.path.join(cachedir, 'lists')
await self.mount(cache_lists, f'/var/lib/apt/lists')
await target.mount(cache_lists, f'/var/lib/apt/lists')
return cachedir
async def pacman_setup(self):
return self.cache_root + '/pacman'
async def pacman_setup(self, target):
cachedir = os.path.join(self.cache_root, self.mgr)
await target.mkdir(cachedir + '/cache', cachedir + '/sync')
await target.mount(cachedir + '/sync', '/var/lib/pacman/sync')
await target.mount(cachedir + '/cache', '/var/cache/pacman')
if await target.host.exists('/etc/pacman.d/mirrorlist'):
await target.copy('/etc/pacman.d/mirrorlist', '/etc/pacman.d/mirrorlist')
def __repr__(self):
return f'Packages({self.packages})'
def __str__(self):
return f'Packages({self.packages}, upgrade={self.upgrade})'

14
shlax/actions/parallel.py Normal file
View File

@ -0,0 +1,14 @@
import asyncio
class Parallel:
def __init__(self, *actions):
self.actions = actions
async def __call__(self, target):
return await asyncio.gather(*[
target(action) for action in self.actions
])
def __str__(self):
return 'Parallel executor'

85
shlax/actions/pip.py
View File

@ -1,57 +1,72 @@
from glob import glob
import os
from urllib import request
from .base import Action
class Pip(Action):
def __init__(self, *pip_packages, pip=None, requirements=None):
self.requirements = requirements
super().__init__(*pip_packages, pip=pip, requirements=requirements)
"""Pip abstraction layer."""
def __init__(self, *pip_packages):
self.pip_packages = pip_packages
async def call(self, *args, **kwargs):
pip = self.kwargs.get('pip', None)
if not pip:
pip = await self.which('pip3', 'pip', 'pip2')
if pip:
pip = pip[0]
async def __call__(self, target):
# ensure python presence
results = await target.which('python3', 'python')
if results:
python = results[0]
else:
from .packages import Packages
action = self.action(
Packages,
'python3,apk', 'python3-pip,apt',
args=args, kwargs=kwargs
raise Exception('Could not find pip nor python')
# ensure pip module presence
result = await target.exec(
python, '-m', 'pip',
raises=False, quiet=True
)
await action(*args, **kwargs)
pip = await self.which('pip3', 'pip', 'pip2')
if not pip:
raise Exception('Could not install a pip command')
else:
pip = pip[0]
if result.rc != 0:
if not os.path.exists('get-pip.py'):
req = request.urlopen(
'https://bootstrap.pypa.io/get-pip.py'
)
content = req.read()
with open('get-pip.py', 'wb+') as f:
f.write(content)
await target.copy('get-pip.py', '.')
await target.exec(python, 'get-pip.py')
# choose a cache directory
if 'CACHE_DIR' in os.environ:
cache = os.path.join(os.getenv('CACHE_DIR'), 'pip')
else:
cache = os.path.join(os.getenv('HOME'), '.cache', 'pip')
if getattr(self, 'mount', None):
# and mount it
if getattr(target, 'mount', None):
# we are in a target which shares a mount command
await self.mount(cache, '/root/.cache/pip')
await self.exec(f'{pip} install --upgrade pip')
await target.mount(cache, '/root/.cache/pip')
# https://github.com/pypa/pip/issues/5599
if 'pip' not in self.kwargs:
pip = 'python3 -m pip'
source = []
nonsource = []
for package in self.pip_packages:
if os.path.exists(package):
source.append(package)
else:
nonsource.append(package)
source = [p for p in self.args if p.startswith('/') or p.startswith('.')]
if source:
await self.exec(
f'{pip} install --upgrade --editable {" ".join(source)}'
if nonsource:
await target.exec(
python, '-m', 'pip',
'install', '--upgrade',
*nonsource
)
nonsource = [p for p in self.args if not p.startswith('/')]
if nonsource:
await self.exec(f'{pip} install --upgrade {" ".join(nonsource)}')
if source:
await target.exec(
python, '-m', 'pip',
'install', '--upgrade', '--editable',
*source
)
if self.requirements:
await self.exec(f'{pip} install --upgrade -r {self.requirements}')
def __str__(self):
return f'Pip({", ".join(self.pip_packages)})'

27
shlax/actions/run.py
View File

@ -1,18 +1,15 @@
from ..targets.buildah import Buildah
from ..targets.docker import Docker
from .base import Action
class Run(Action):
async def call(self, *args, **kwargs):
image = self.kwargs.get('image', None)
if not image:
return await self.exec(*self.args, **self.kwargs)
if isinstance(image, Buildah):
breakpoint()
result = await self.action(image, *args, **kwargs)
class Run:
def __init__(self, cmd, root=False):
self.cmd = cmd
self.root = root
return await Docker(
image=image,
).exec(*args, **kwargs)
async def __call__(self, target):
if self.root:
self.proc = await target.rexec(self.cmd)
else:
self.proc = await target.exec(self.cmd)
def __str__(self):
return f'Run({self.cmd})'

16
shlax/actions/service.py
View File

@ -1,16 +0,0 @@
import asyncio
from .base import Action
class Service(Action):
def __init__(self, *names, state=None):
self.state = state or 'started'
self.names = names
super().__init__()
async def call(self, *args, **kwargs):
return asyncio.gather(*[
self.exec('systemctl', 'start', name, user='root')
for name in self.names
])

44
shlax/actions/user.py Normal file
View File

@ -0,0 +1,44 @@
import os
import re
from .packages import Packages
class User:
"""
Create a user.
Example:
User('app', '/app', getenv('_CONTAINERS_ROOTLESS_UID', 1000)),
_CONTAINERS_ROOTLESS_UID allows to get your UID during build, which happens
in buildah unshare.
"""
def __init__(self, username, home, uid):
self.username = username
self.home = home
self.uid = uid
def __str__(self):
return f'User({self.username}, {self.home}, {self.uid})'
async def __call__(self, target):
result = await target.rexec('id', self.uid, raises=False)
if result.rc == 0:
old = re.match('.*\(([^)]*)\).*', result.out).group(1)
await target.rexec(
'usermod',
'-d', self.home,
'-l', self.username,
old
)
else:
await target.rexec(
'useradd',
'-d', self.home,
'-u', self.uid,
self.username
)
await target.mkdir(self.home)
await target.rexec('chown', self.uid, self.home)

228
shlax/cli.py
View File

@ -1,146 +1,124 @@
'''
shlax is a micro-framework to orchestrate commands.
shlax yourfile.py: to list actions you have declared.
shlax yourfile.py <action>: to execute a given action
#!/usr/bin/env shlax: when making yourfile.py an executable.
'''
"""
Shlax executes mostly in 3 ways:
- Execute actions on targets with the command line
- With your shlaxfile as first argument: offer defined Actions
- With the name of a module in shlax.repo: a community maintained shlaxfile
"""
import ast
import asyncio
import cli2
import copy
import glob
import inspect
import importlib
import os
import sys
from .exceptions import *
from .shlaxfile import Shlaxfile
from .targets import Localhost
from .proc import ProcFailure
async def runall(*args, **kwargs):
for name, action in cli.shlaxfile.actions.items():
await Localhost(action)(*args, **kwargs)
class Group(cli2.Group):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cmdclass = Command
@cli2.option('debug', alias='d', help='Display debug output.')
async def test(*args, **kwargs):
"""Run podctl test over a bunch of paths."""
report = []
class TargetArgument(cli2.Argument):
"""
Target to execute on: localhost by default, target=@ssh_host for ssh.
"""
for arg in args:
candidates = [
os.path.join(os.getcwd(), arg, 'pod.py'),
os.path.join(os.getcwd(), arg, 'pod_test.py'),
]
for candidate in candidates:
if not os.path.exists(candidate):
continue
podfile = Podfile.factory(candidate)
def __init__(self, cmd, param, doc=None, color=None, default=None):
from shlax.targets.base import Target
super().__init__(cmd, param, doc=self.__doc__, default=Target())
self.alias = ['target', 't']
# disable push
for name, container in podfile.containers.items():
commit = container.visitor('commit')
if commit:
commit.push = False
def cast(self, value):
from shlax.targets.ssh import Ssh
user, host = value.split('@')
return Ssh(host=host, user=user)
output.print(
'\n\x1b[1;38;5;160;48;5;118m BUILD START \x1b[0m'
+ ' ' + podfile.path + '\n'
def match(self, arg):
return arg if isinstance(arg, str) and '@' in arg else None
class Command(cli2.Command):
def setargs(self):
super().setargs()
if 'target' in self.sig.parameters:
self['target'] = TargetArgument(
self,
self.sig.parameters['target'],
)
if 'actions' in self:
del self['actions']
old_exit_code = console_script.exit_code
console_script.exit_code = 0
try:
await podfile.pod.script('build')()
except Exception as e:
report.append(('build ' + candidate, False))
continue
if console_script.exit_code != 0:
report.append(('build ' + candidate, False))
continue
console_script.exit_code = old_exit_code
for name, test in podfile.tests.items():
name = '::'.join([podfile.path, name])
output.print(
'\n\x1b[1;38;5;160;48;5;118m TEST START \x1b[0m'
+ ' ' + name + '\n'
)
def __call__(self, *argv):
result = None
try:
await test(podfile.pod)
except Exception as e:
report.append((name, False))
output.print('\x1b[1;38;5;15;48;5;196m TEST FAIL \x1b[0m' + name)
else:
report.append((name, True))
output.print('\x1b[1;38;5;200;48;5;44m TEST SUCCESS \x1b[0m' + name)
output.print('\n')
result = super().__call__(*argv)
except ProcFailure:
# just output the failure without TB, as command was already
# printed anyway
pass
print('\n')
for name, success in report:
if success:
output.print('\n\x1b[1;38;5;200;48;5;44m TEST SUCCESS \x1b[0m' + name)
else:
output.print('\n\x1b[1;38;5;15;48;5;196m TEST FAIL \x1b[0m' + name)
print('\n')
success = [*filter(lambda i: i[1], report)]
failures = [*filter(lambda i: not i[1], report)]
output.print(
'\n\x1b[1;38;5;200;48;5;44m TEST TOTAL: \x1b[0m'
+ str(len(report))
)
if success:
output.print(
'\n\x1b[1;38;5;200;48;5;44m TEST SUCCESS: \x1b[0m'
+ str(len(success))
)
if failures:
output.print(
'\n\x1b[1;38;5;15;48;5;196m TEST FAIL: \x1b[0m'
+ str(len(failures))
)
if failures:
console_script.exit_code = 1
class ConsoleScript(cli2.ConsoleScript):
def __call__(self, *args, **kwargs):
self.shlaxfile = None
shlaxfile = sys.argv.pop(1) if len(sys.argv) > 1 else ''
if os.path.exists(shlaxfile.split('::')[0]):
self.shlaxfile = Shlaxfile()
self.shlaxfile.parse(shlaxfile)
for name, action in self.shlaxfile.actions.items():
self[name] = cli2.Callable(
name,
action.callable(),
options={
k: cli2.Option(name=k, **v)
for k, v in action.options.items()
},
color=getattr(action, 'color', cli2.YELLOW),
)
return super().__call__(*args, **kwargs)
def call(self, command):
kwargs = copy.copy(self.parser.funckwargs)
kwargs.update(self.parser.options)
try:
return command(*self.parser.funcargs, **kwargs)
except WrongResult as e:
print(e)
self.exit_code = e.proc.rc
except ShlaxException as e:
print(e)
if self['target'].value.results:
if self['target'].value.results[-1].status == 'failure':
self.exit_code = 1
self['target'].value.output.results(self['target'].value)
return result
cli = ConsoleScript(__doc__).add_module('shlax.cli')
class ActionCommand(cli2.Command):
def setargs(self):
super().setargs()
self['target'] = TargetArgument(
self,
inspect.Parameter('target', inspect.Parameter.KEYWORD_ONLY),
)
def call(self, *args, **kwargs):
self.target = self.target(*args, **kwargs)
return super().call(self['target'].value)
class ConsoleScript(Group):
def __call__(self, *argv):
self.load_actions()
#self.load_shlaxfiles() # wip
return super().__call__(*argv)
def load_shlaxfiles(self):
filesdir = os.path.dirname(__file__) + '/shlaxfiles/'
for filename in os.listdir(filesdir):
filepath = filesdir + filename
if not os.path.isfile(filepath):
continue
with open(filepath, 'r') as f:
tree = ast.parse(f.read())
group = self.group(filename[:-3])
main = Group(doc=__doc__).load(shlax)
def load_actions(self):
actionsdir = os.path.dirname(__file__) + '/actions/'
for filename in os.listdir(actionsdir):
filepath = actionsdir + filename
if not os.path.isfile(filepath):
continue
with open(filepath, 'r') as f:
tree = ast.parse(f.read())
cls = [
node
for node in tree.body
if isinstance(node, ast.ClassDef)
]
if not cls:
continue
mod = importlib.import_module('shlax.actions.' + filename[:-3])
cls = getattr(mod, cls[0].name)
self.add(cls, name=filename[:-3], cmdclass=ActionCommand)
cli = ConsoleScript(doc=__doc__)

129
shlax/container.py Normal file
View File

@ -0,0 +1,129 @@
import copy
import os
from .podman import Podman
from .image import Image
class Container:
def __init__(self, build=None, image=None, env=None, volumes=None):
self.build = build
self.image = image or self.build.image
if isinstance(self.image, str):
self.image = Image(self.image)
self.volumes = volumes or {}
self.env = env or {}
prefix = os.getcwd().split('/')[-1]
repo = self.image.repository.replace('/', '-')
if prefix == repo:
self.name = repo
else:
self.name = '-'.join([prefix, repo])
self.pod = None
@property
def full_name(self):
if self.pod:
return '-'.join([self.pod.name, self.name])
return self.name
async def up(self, target, *args):
"""Start the container foreground"""
podman = Podman(target)
if self.pod:
pod = None
for _ in await podman.pod.ps():
if _['Name'] == self.pod.name:
pod = _
break
if not pod:
await podman.pod.create('--name', self.pod.name)
args = list(args) + ['--pod', self.pod.name]
# skip if already up
for result in await podman.ps('-a'):
for name in result['Names']:
if name == self.full_name:
if result['State'] == 'running':
target.output.info(f'{self.full_name} already running')
return
elif result['State'] in ('exited', 'configured'):
target.output.info(f'{self.full_name} starting')
startargs = ['podman', 'start']
if '-d' not in args:
startargs.append('--attach')
startargs.append(self.full_name)
await target.exec(*startargs)
return
cmd = [
'podman',
'run',
] + list(args)
for src, dest in self.volumes.items():
cmd += ['--volume', ':'.join([src, dest])]
for src, dest in self.env.items():
cmd += ['--env', '='.join([src, str(dest)])]
cmd += [
'--name',
self.full_name,
str(self.image),
]
await target.exec(*cmd)
async def start(self, target):
"""Start the container background"""
await self.up(target, '-d')
async def stop(self, target):
"""Start the container"""
await target.exec('podman', 'stop', self.full_name)
async def inspect(self, target):
"""Inspect container"""
await target.exec('podman', 'inspect', self.full_name)
async def logs(self, target):
"""Show container logs"""
await target.exec('podman', 'logs', self.full_name)
async def exec(self, target, cmd=None):
"""Execute a command in the container"""
cmd = cmd or 'bash'
if cmd.endswith('sh'):
import os
os.execvp(
'/usr/bin/podman',
[
'podman',
'exec',
'-it',
self.full_name,
cmd,
]
)
result = await target.exec(
'podman',
'exec',
self.full_name,
cmd,
)
async def down(self, target):
"""Start the container"""
await target.exec('podman', 'rm', '-f', self.full_name, raises=False)
async def apply(self, target):
"""Start the container"""
if self.build:
await target(self.build)
await target(self.down)
await target(self.start)
def __str__(self):
return f'Container(name={self.name}, image={self.image}, volumes={self.volumes})'

0
shlax/contrib/__init__.py Normal file
View File

37
shlax/contrib/gitlab.py
View File

@ -1,37 +0,0 @@
from copy import deepcopy
import yaml
from shlax import *
class GitLabCI(Script):
async def call(self, *args, write=True, **kwargs):
output = dict()
for key, value in self.kwargs.items():
if isinstance(value, dict):
output[key] = deepcopy(value)
image = output[key].get('image', 'alpine')
if hasattr(image, 'image'):
output[key]['image'] = image.image.repository + ':$CI_COMMIT_SHORT_SHA'
else:
output[key] = value
output = yaml.dump(output)
if kwargs['debug'] is True:
self.output(output)
if write:
with open('.gitlab-ci.yml', 'w+') as f:
f.write(output)
from shlax.cli import cli
for arg in args:
job = self.kwargs[arg]
_args = []
if not isinstance(job['image'], str):
image = str(job['image'].image)
else:
image = job['image']
await self.action('Docker', Run(job['script']), image=image)(*_args, **kwargs)
def colorized(self):
return type(self).__name__

22
shlax/exceptions.py
View File

@ -1,22 +0,0 @@
class ShlaxException(Exception):
pass
class Mistake(ShlaxException):
pass
class WrongResult(ShlaxException):
def __init__(self, proc):
self.proc = proc
msg = f'FAIL exit with {proc.rc} ' + proc.args[0]
if not proc.debug or 'cmd' not in str(proc.debug):
msg += '\n' + proc.cmd
if not proc.debug or 'out' not in str(proc.debug):
msg += '\n' + proc.out
msg += '\n' + proc.err
super().__init__(msg)

81
shlax/image.py
View File

@ -1,31 +1,52 @@
import json
import os
import re
class Image:
ENV_TAGS = (
# gitlab
'CI_COMMIT_SHORT_SHA',
'CI_COMMIT_REF_NAME',
'CI_COMMIT_TAG',
# CircleCI
'CIRCLE_SHA1',
'CIRCLE_TAG',
'CIRCLE_BRANCH',
# contributions welcome here
)
class Layers(set):
def __init__(self, image):
self.image = image
async def ls(self, target):
"""Fetch layers from localhost"""
ret = set()
results = await target.parent.exec(
'buildah images --json',
quiet=True,
)
results = json.loads(results.out)
prefix = 'localhost/' + self.image.repository + ':layer-'
for result in results:
if not result.get('names', None):
continue
for name in result['names']:
if name.startswith(prefix):
self.add(name)
return self
async def rm(self, target, tags=None):
"""Drop layers for this image"""
if tags is None:
tags = [layer for layer in await self.ls(target)]
await target.exec('podman', 'rmi', *tags, raises=False)
class Image:
PATTERN = re.compile(
'^((?P<backend>[a-z]*)://)?((?P<registry>[^/]*[.][^/]*)/)?((?P<repository>[^:]+))?(:(?P<tags>.*))?$' # noqa
, re.I
)
def __init__(self, arg=None, format=None, backend=None, registry=None, repository=None, tags=None):
def __init__(self, arg=None, format=None, backend=None, registry=None,
repository=None, tags=None):
self.arg = arg
self.format = format
self.backend = backend
self.registry = registry
self.repository = repository
self.tags = tags or []
self.layers = Layers(self)
match = re.match(self.PATTERN, arg)
if match:
@ -39,15 +60,11 @@ class Image:
setattr(self, k, v)
# docker.io currently has issues with oci format
self.format = format or 'oci'
if self.registry == 'docker.io':
self.format = 'docker'
self.backend = 'docker'
# figure tags from CI vars
for name in self.ENV_TAGS:
value = os.getenv(name)
if value:
self.tags.append(value)
if not self.format:
self.format = 'docker' if self.backend == 'docker' else 'oci'
# filter out tags which resolved to None
self.tags = [t for t in self.tags if t]
@ -56,20 +73,22 @@ class Image:
if not self.tags:
self.tags = ['latest']
async def __call__(self, action, *args, **kwargs):
args = list(args)
return await action.exec(*args, **self.kwargs)
def __str__(self):
return f'{self.repository}:{self.tags[-1]}'
async def push(self, *args, **kwargs):
user = os.getenv('DOCKER_USER')
passwd = os.getenv('DOCKER_PASS')
action = kwargs.get('action', self)
async def push(self, target, name=None):
user = os.getenv('IMAGES_USER', os.getenv('DOCKER_USER'))
passwd = os.getenv('IMAGES_PASS', os.getenv('DOCKER_PASS'))
if user and passwd:
action.output.cmd('buildah login -u ... -p ...' + self.registry)
await action.exec('buildah', 'login', '-u', user, '-p', passwd, self.registry or 'docker.io', debug=False)
target.output.cmd('buildah login -u ... -p ...' + self.registry)
await target.parent.exec(
'buildah', 'login', '-u', user, '-p', passwd,
self.registry or 'docker.io', quiet=True)
for tag in self.tags:
await action.exec('buildah', 'push', f'{self.repository}:{tag}')
await target.parent.exec(
'buildah',
'push',
self.repository + ':final',
name if isinstance(name, str) else f'{self.registry}/{self.repository}:{tag}'
)

84
shlax/output.py
View File

@ -1,5 +1,6 @@
import re
import sys
import types
from .colors import colors
@ -25,7 +26,23 @@ class Output:
def colorize(self, code, content):
return self.color(code) + content + self.color()
def __init__(self, prefix=None, regexps=None, debug=False, write=None, flush=None, **kwargs):
def colorized(self, action):
if hasattr(action, 'colorized'):
return action.colorized(self.colors)
elif isinstance(action, types.MethodType):
return f'{action.__self__}.{action.__name__}'
else:
return str(action)
def __init__(
self,
prefix=None,
regexps=None,
debug='cmd,visit,out',
write=None,
flush=None,
**kwargs
):
self.prefix = prefix
self.debug = debug
self.prefix_length = 0
@ -84,7 +101,11 @@ class Output:
)
def highlight(self, line, highlight=True):
try:
line = line.decode('utf8') if isinstance(line, bytes) else line
except UnicodeDecodeError:
highlight = False
if not highlight or (
'\x1b[' in line
or '\033[' in line
@ -99,22 +120,21 @@ class Output:
return line
def test(self, action):
if self.debug is True:
self(''.join([
self.colors['purplebold'],
'! TEST ',
self.colors['reset'],
action.colorized(),
self.colorized(action),
'\n',
]))
def clean(self, action):
if self.debug is True:
if self.debug:
self(''.join([
self.colors['bluebold'],
'+ CLEAN ',
self.colors['reset'],
action.colorized(),
self.colorized(action),
'\n',
]))
@ -124,7 +144,27 @@ class Output:
self.colors['orangebold'],
'⚠ START ',
self.colors['reset'],
action.colorized(),
self.colorized(action),
'\n',
]))
def info(self, text):
if self.debug is True or 'visit' in str(self.debug):
self(''.join([
self.colors['cyanbold'],
'➤ INFO ',
self.colors['reset'],
text,
'\n',
]))
def skip(self, action):
if self.debug is True or 'visit' in str(self.debug):
self(''.join([
self.colors['yellowbold'],
'↪️ SKIP ',
self.colors['reset'],
self.colorized(action),
'\n',
]))
@ -134,7 +174,7 @@ class Output:
self.colors['greenbold'],
'✔ SUCCESS ',
self.colors['reset'],
action.colorized() if hasattr(action, 'colorized') else str(action),
self.colorized(action),
'\n',
]))
@ -144,6 +184,34 @@ class Output:
self.colors['redbold'],
'✘ FAIL ',
self.colors['reset'],
action.colorized() if hasattr(action, 'colorized') else str(action),
self.colorized(action),
'\n',
]))
def results(self, action):
if len(action.results) < 2:
return
success = 0
fail = 0
for result in action.results:
if result.status == 'success':
success += 1
if result.status == 'failure':
fail += 1
self(''.join([
self.colors['greenbold'],
'✔ SUCCESS REPORT: ',
self.colors['reset'],
str(success),
'\n',
]))
if fail:
self(''.join([
self.colors['redbold'],
'✘ FAIL REPORT: ',
self.colors['reset'],
str(fail),
'\n',
]))

78
shlax/pod.py Normal file
View File

@ -0,0 +1,78 @@
import cli2
import json
import os
import sys
from shlax.targets.base import Target
from shlax.actions.parallel import Parallel
from shlax.proc import Proc
from .podman import Podman
class Pod:
"""Help text"""
def __init__(self, **containers):
self.containers = containers
for name, container in self.containers.items():
container.pod = self
container.name = name
self.name = os.getcwd().split('/')[-1]
async def _call(self, target, method, *names):
methods = [
getattr(container, method)
for name, container in self.containers.items()
if not names or name in names
]
await target(Parallel(*methods))
async def build(self, target, *names):
"""Build container images"""
if not (Proc.test or os.getuid() == 0):
os.execvp('buildah', ['buildah', 'unshare'] + sys.argv)
else:
await self._call(target, 'build', *names)
async def down(self, target, *names):
"""Delete container images"""
await self._call(target, 'down', *names)
async def start(self, target, *names):
"""Start container images"""
await self._call(target, 'start', *names)
async def logs(self, target, *names):
"""Start container images"""
await self._call(target, 'logs', *names)
async def ps(self, target):
"""Show containers and volumes"""
containers = []
names = []
for container in await Podman(target).ps('-a'):
for name in container['Names']:
if name.startswith(self.name + '-'):
container['Name'] = name
containers.append(container)
names.append(name)
for name, container in self.containers.items():
full_name = '-'.join([self.name, container.name])
if full_name in names:
continue
containers.append(dict(
Name=full_name,
State='not created',
))
cli2.Table(
['Name', 'State'],
*[
(container['Name'], container['State'])
for container in containers
]
).print()
def __str__(self):
return f'Pod({self.name})'

20
shlax/podman.py Normal file
View File

@ -0,0 +1,20 @@
import json
class Podman(list):
def __init__(self, target, *args):
self.target = target
super().__init__(args or ['podman'])
def __getattr__(self, command):
if command.startswith('_'):
return super().__getattr__(command)
return Podman(self.target, *self + [command])
async def __call__(self, *args, **kwargs):
cmd = self + list(args) + [
f'--{k}={v}' for k, v in kwargs.items()
]
if 'ps' in cmd:
cmd += ['--format=json']
return (await self.target.exec(*cmd, quiet=True)).json

41
shlax/proc.py
View File

@ -7,31 +7,46 @@ import os
import shlex
import sys
from .exceptions import WrongResult
from .output import Output
class ProcFailure(Exception):
def __init__(self, proc):
self.proc = proc
msg = f'FAIL exit with {proc.rc} ' + proc.args[0]
if not proc.output.debug or 'cmd' not in str(proc.output.debug):
msg += '\n' + proc.cmd
if not proc.output.debug or 'out' not in str(proc.output.debug):
msg += '\n' + proc.out
msg += '\n' + proc.err
super().__init__(msg)
class PrefixStreamProtocol(asyncio.subprocess.SubprocessStreamProtocol):
"""
Internal subprocess stream protocol to add a prefix in front of output to
make asynchronous output readable.
"""
def __init__(self, output, *args, **kwargs):
self.output = output
def __init__(self, proc, *args, **kwargs):
self.proc = proc
super().__init__(*args, **kwargs)
def pipe_data_received(self, fd, data):
if self.output.debug is True or 'out' in str(self.output.debug):
if self.proc.output.debug is True or 'out' in str(self.proc.output.debug):
if fd in (1, 2):
self.output(data)
self.proc.output(data)
super().pipe_data_received(fd, data)
def protocol_factory(output):
def protocol_factory(proc):
def _p():
return PrefixStreamProtocol(
output,
proc,
limit=asyncio.streams._DEFAULT_LIMIT,
loop=asyncio.events.get_event_loop()
)
@ -54,8 +69,10 @@ class Proc:
"""
test = False
def __init__(self, *args, prefix=None, raises=True, debug=None, output=None):
self.debug = debug if not self.test else False
def __init__(self, *args, prefix=None, raises=True, output=None, quiet=False):
if quiet:
self.output = Output(debug=False)
else:
self.output = output or Output()
self.cmd = ' '.join(args)
self.args = args
@ -87,7 +104,7 @@ class Proc:
if self.called:
raise Exception('Already called: ' + self.cmd)
if self.debug is True or 'cmd' in str(self.debug):
if 'cmd' in str(self.output.debug):
self.output.cmd(self.cmd)
if self.test:
@ -98,7 +115,7 @@ class Proc:
loop = asyncio.events.get_event_loop()
transport, protocol = await loop.subprocess_exec(
protocol_factory(self.output), *self.args)
protocol_factory(self), *self.args)
self.proc = asyncio.subprocess.Process(transport, protocol, loop)
self.called = True
@ -123,7 +140,7 @@ class Proc:
if not self.communicated:
await self.communicate()
if self.raises and self.proc.returncode:
raise WrongResult(self)
raise ProcFailure(self)
return self
@property

0
shlax/repo/__init__.py Normal file
View File

13
shlax/result.py Normal file
View File

@ -0,0 +1,13 @@
class Result:
def __init__(self, target, action):
self.target = target
self.action = action
self.status = 'pending'
self.exception = None
class Results(list):
def new(self, target, action):
result = Result(target, action)
self.append(result)
return result

27
shlax/shlaxfile.py
View File

@ -1,27 +0,0 @@
import importlib
import os
from .actions.base import Action
class Shlaxfile:
def __init__(self, actions=None, tests=None):
self.actions = actions or {}
self.tests = tests or {}
self.paths = []
def parse(self, path):
spec = importlib.util.spec_from_file_location('shlaxfile', path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
for name, value in mod.__dict__.items():
if isinstance(value, Action):
value.name = name
self.actions[name] = value
elif callable(value) and getattr(value, '__name__', '').startswith('test_'):
self.tests[value.__name__] = value
self.paths.append(path)
@property
def path(self):
return self.paths[0]

18
shlax/shortcuts.py Normal file
View File

@ -0,0 +1,18 @@
from .targets.base import Target
from .targets.buildah import Buildah
from .targets.localhost import Localhost
from .targets.stub import Stub
from .actions.copy import Copy
from .actions.packages import Packages
from .actions.run import Run
from .actions.pip import Pip
from .actions.parallel import Parallel
from .actions.user import User
from .cli import Command, Group
from .container import Container
from .pod import Pod
from os import getenv, environ

3
shlax/strategies/__init__.py
View File

@ -1,3 +0,0 @@
from .asyn import Async
from .script import Script
from .pod import Pod, Container

11
shlax/strategies/asyn.py
View File

@ -1,11 +0,0 @@
import asyncio
from .script import Script
class Async(Script):
async def call(self, *args, **kwargs):
return await asyncio.gather(*[
action(*args, **kwargs)
for action in self.actions
])

27
shlax/strategies/pod.py
View File

@ -1,27 +0,0 @@
import os
from .script import Script
class Container(Script):
async def call(self, *args, **kwargs):
if not args or 'build' in args:
await self.kwargs['build'](**kwargs)
self.image = self.kwargs['build'].image
if not args or 'test' in args:
self.output.test(self)
await self.action('Docker',
*self.kwargs['test'].actions,
image=self.image,
mount={'.': '/app'},
workdir='/app',
)(**kwargs)
if not args or 'push' in args:
await self.image.push(action=self)
#name = kwargs.get('name', os.getcwd()).split('/')[-1]
class Pod(Script):
pass

34
shlax/strategies/script.py
View File

@ -1,34 +0,0 @@
import copy
import os
from ..exceptions import WrongResult
from ..actions.base import Action
from ..proc import Proc
class Actions(list):
def __init__(self, owner, actions):
self.owner = owner
super().__init__()
for action in actions:
self.append(action)
def append(self, value):
value = copy.deepcopy(value)
value.parent = self.owner
value.status = 'pending'
super().append(value)
class Script(Action):
contextualize = ['shargs', 'exec', 'rexec', 'env', 'which', 'copy']
def __init__(self, *actions, **kwargs):
super().__init__(**kwargs)
self.actions = Actions(self, actions)
async def call(self, *args, **kwargs):
for action in self.actions:
result = await action(*args, **kwargs)
if action.status != 'success':
break

4
shlax/targets/__init__.py
View File

@ -1,4 +0,0 @@
from .buildah import Buildah
from .docker import Docker
from .localhost import Localhost
from .ssh import Ssh

201
shlax/targets/base.py Normal file
View File

@ -0,0 +1,201 @@
import asyncio
import copy
from pathlib import Path
import os
import re
import sys
from ..output import Output
from ..proc import Proc, ProcFailure
from ..result import Result, Results
class Target:
isguest = False
def __init__(self, *actions, root=None):
self.actions = actions
self.results = []
self.output = Output()
self.parent = None
self.root = root or ''
def __str__(self):
return 'localhost'
@property
def parent(self):
return self._parent or Target()
@parent.setter
def parent(self, value):
self._parent = value
@property
def caller(self):
"""Traverse parents and return the top-levels Target."""
if not self._parent:
return self
caller = self._parent
while caller._parent:
caller = caller._parent
return caller
async def __call__(self, *actions, target=None):
if target:
# that's going to be used by other target methods, to access
# the calling target
self.parent = target
result = Result(self, self)
result.status = 'success'
for action in actions or self.actions:
if await self.action(action, reraise=bool(actions)):
result.status = 'failure'
break
if getattr(self, 'clean', None):
self.output.clean(self)
await self.clean(self, result)
async def action(self, action, reraise=False):
result = Result(self, action)
self.output.start(action)
try:
await action(target=self)
except Exception as e:
self.output.fail(action, e)
result.status = 'failure'
result.exception = e
if not isinstance(e, ProcFailure):
# no need to reraise in case of command error
# because the command has been printed
if reraise:
# nested call, re-raise
raise
else:
import traceback
traceback.print_exception(type(e), e, sys.exc_info()[2])
return True # because it failed
else:
if getattr(action, 'skipped', False):
self.output.skip(action)
else:
self.output.success(action)
result.status = 'success'
finally:
self.caller.results.append(result)
clean = getattr(action, 'clean', None)
if clean:
self.output.clean(action)
await clean(self, result)
async def rexec(self, *args, **kwargs):
kwargs['user'] = 'root'
return await self.exec(*args, **kwargs)
async def which(self, *cmd):
"""
Return the first path to the cmd in the container.
If cmd argument is a list then it will try all commands.
"""
proc = await self.exec('type ' + ' '.join(cmd), raises=False)
result = []
for res in proc.out.split('\n'):
match = re.match('([^ ]+) is ([^ ]+)$', res.strip())
if match:
result.append(match.group(1))
return result
def shargs(self, *args, **kwargs):
user = kwargs.pop('user', None)
args = [str(arg) for arg in args if args is not None]
if args and ' ' in args[0]:
if len(args) == 1:
args = ['sh', '-euc', args[0]]
else:
args = ['sh', '-euc'] + list(args)
if user == 'root':
args = ['sudo'] + args
elif user:
args = ['sudo', '-u', user] + args
return args, kwargs
if self.parent:
return self.parent.shargs(*args, **kwargs)
else:
return args, kwargs
async def exec(self, *args, **kwargs):
kwargs['output'] = self.output
args, kwargs = self.shargs(*args, **kwargs)
proc = await Proc(*args, **kwargs)()
if kwargs.get('wait', True):
await proc.wait()
return proc
@property
def root(self):
return self._root
@root.setter
def root(self, value):
self._root = Path(value) if value else ''
@property
def host(self):
current = self
while current.isguest:
current = self.parent
return current
def path(self, path):
if not self.root:
return path
if str(path).startswith('/'):
path = str(path)[1:]
return str(self.root / path)
async def mkdir(self, *paths):
if '_mkdir' not in self.__dict__:
self._mkdir = []
make = [str(path) for path in paths if str(path) not in self._mkdir]
if make:
await self.exec('mkdir', '-p', *make)
self._mkdir += make
async def copy(self, *args):
return await self.exec('cp', '-a', *args)
async def exists(self, path):
return (await self.exec('ls ' + self.path(path), raises=False)).rc == 0
async def read(self, path):
return (await self.exec('cat', self.path(path))).out
async def write(self, path, content, **kwargs):
return await self.exec(
f'cat > {self.path(path)} <<EOF\n'
+ content
+ '\nEOF',
**kwargs
)
async def rm(self, path):
return await self.exec('rm', self.path(path))
async def getenv(self, key):
return (await self.exec('echo $' + key)).out
async def getcwd(self):
return (await self.exec('pwd')).out

325
shlax/targets/buildah.py
View File

@ -1,156 +1,231 @@
import asyncio
import copy
import hashlib
import json
import os
import asyncio
from pathlib import Path
import signal
import shlex
import subprocess
import sys
import textwrap
from pathlib import Path
from .base import Target
from ..actions.base import Action
from ..exceptions import Mistake
from ..proc import Proc
from ..image import Image
from .localhost import Localhost
from ..proc import Proc
class Buildah(Localhost):
"""
The build script iterates over visitors and runs the build functions, it
also provides wrappers around the buildah command.
"""
contextualize = Localhost.contextualize + ['mnt', 'ctr', 'mount', 'image']
class Buildah(Target):
"""Build container image with buildah"""
isguest = True
def __init__(self, base, *args, commit=None, push=False, cmd=None, **kwargs):
if isinstance(base, Action):
args = [base] + list(args)
base = 'alpine' # default selection in case of mistake
super().__init__(*args, **kwargs)
self.base = base
self.mounts = dict()
self.ctr = None
self.mnt = None
def __init__(self, *actions, base=None, commit=None):
self.base = base or 'alpine'
self.image = Image(commit) if commit else None
self.config= dict(
cmd=cmd or 'sh',
self.ctr = None
self.root = None
self.mounts = dict()
# Always consider localhost as parent for now
self.parent = Target()
super().__init__(*actions)
def is_runnable(self):
return Proc.test or os.getuid() == 0
def __str__(self):
if not self.is_runnable():
return 'Replacing with: buildah unshare ' + ' '.join(sys.argv)
return f'Buildah({self.image})'
async def __call__(self, *actions, target=None, push: str=False):
if target:
self.parent = target
self.push = push
if not self.is_runnable():
os.execvp('buildah', ['buildah', 'unshare'] + sys.argv)
return # process has been replaced
layers = await self.image.layers.ls(self)
keep = await self.cache_setup(self.image.layers, *actions)
keepnames = [*map(lambda x: 'localhost/' + str(x), keep)]
self.invalidate = [name for name in self.image.layers if name not in keepnames]
if self.invalidate:
self.output.info('Invalidating old layers')
await self.image.layers.rm(self.parent, self.invalidate)
if actions:
actions = actions[len(keep):]
else:
self.actions = self.actions[len(keep):]
self.ctr = (await self.parent.exec('buildah', 'from', self.base)).out
self.root = Path((await self.parent.exec('buildah', 'mount', self.ctr)).out)
return await super().__call__(*actions)
async def cache_setup(self, layers, *actions):
keep = []
self.image_previous = Image(self.base)
for action in actions or self.actions:
action_image = await self.action_image(action)
name = 'localhost/' + str(action_image)
if name in layers:
self.base = self.image_previous = action_image
keep.append(action_image)
self.output.skip(
f'Found layer for {action}: {action_image.tags[0]}'
)
else:
break
return keep
def shargs(self, *args, user=None, buildah=True, **kwargs):
if not buildah or args[0].startswith('buildah'):
return super().shargs(*args, user=user, **kwargs)
async def action_image(self, action):
prefix = str(self.image_previous)
for tag in self.image_previous.tags:
if tag.startswith('layer-'):
prefix = tag
break
if hasattr(action, 'cachekey'):
action_key = action.cachekey()
if asyncio.iscoroutine(action_key):
action_key = str(await action_key)
else:
action_key = str(action)
key = prefix + action_key
sha1 = hashlib.sha1(key.encode('ascii'))
action_image = copy.deepcopy(self.image)
action_image.tags = ['layer-' + sha1.hexdigest()]
return action_image
async def action(self, action, reraise=False):
stop = await super().action(action, reraise)
if not stop:
action_image = await self.action_image(action)
self.output.info(f'Commiting {action_image} for {action}')
await self.parent.exec(
'buildah',
'commit',
self.ctr,
action_image,
)
self.image_previous = action_image
return stop
async def clean(self, target, result):
if self.ctr is not None:
for src, dst in self.mounts.items():
await self.parent.exec('umount', self.root / str(dst)[1:])
await self.parent.exec('buildah', 'umount', self.ctr)
if result.status == 'success' and self.ctr:
await self.commit()
if self.push:
await self.image.push(target, self.push)
if self.ctr is not None:
await self.parent.exec('buildah', 'rm', self.ctr)
async def mount(self, src, dst):
"""Mount a host directory into the container."""
target = self.root / str(dst)[1:]
await self.parent.exec(f'mkdir -p {src} {target}')
await self.parent.exec(f'mount -o bind {src} {target}')
self.mounts[src] = dst
async def exec(self, *args, user=None, **kwargs):
_args = ['buildah', 'run']
if user:
_args += ['--user', user]
_args += [self.ctr, '--', 'sh', '-euc']
return super().shargs(
*(
_args
+ [' '.join([str(a) for a in args])]
),
**kwargs
)
def __repr__(self):
return f'Base({self.base})'
async def config(self, line):
"""Run buildah config."""
return await self.exec(f'buildah config {line} {self.ctr}', buildah=False)
async def copy(self, *args):
"""Run buildah copy to copy a file from host into container."""
src = args[:-1]
dst = args[-1]
await self.mkdir(dst)
procs = []
for s in src:
if Path(s).is_dir():
target = self.mnt / s
if not target.exists():
await self.mkdir(target)
args = ['buildah', 'copy', self.ctr, s, Path(dst) / s]
else:
args = ['buildah', 'copy', self.ctr, s, dst]
procs.append(self.exec(*args, buildah=False))
return await asyncio.gather(*procs)
async def mount(self, src, dst):
"""Mount a host directory into the container."""
target = self.mnt / str(dst)[1:]
await self.exec(f'mkdir -p {src} {target}', buildah=False)
await self.exec(f'mount -o bind {src} {target}', buildah=False)
self.mounts[src] = dst
def is_runnable(self):
return (
Proc.test
or os.getuid() == 0
)
async def call(self, *args, **kwargs):
if self.is_runnable():
self.ctr = (await self.exec('buildah', 'from', self.base, buildah=False)).out
self.mnt = Path((await self.exec('buildah', 'mount', self.ctr, buildah=False)).out)
result = await super().call(*args, **kwargs)
return result
from shlax.cli import cli
debug = kwargs.get('debug', False)
# restart under buildah unshare environment
argv = [
'buildah', 'unshare',
sys.argv[0], # current script location
cli.shlaxfile.path, # current shlaxfile location
]
if debug is True:
argv.append('-d')
elif isinstance(debug, str) and debug:
argv.append('-d=' + debug)
argv += [
cli.parser.command.name, # script name ?
]
await self.exec(*argv)
_args += [' '.join([str(a) for a in args])]
return await self.parent.exec(*_args, **kwargs)
async def commit(self):
if not self.image:
return
for key, value in self.config.items():
await self.exec(f'buildah config --{key} "{value}" {self.ctr}')
self.sha = (await self.exec(
await self.parent.exec(
'buildah',
'commit',
'--format=' + self.image.format,
f'--format={self.image.format}',
self.ctr,
buildah=False,
)).out
f'{self.image.repository}:final',
)
if self.image.backend == 'docker':
await self.parent.exec(
'buildah',
'push',
f'{self.image.repository}:final',
f'docker-daemon:{self.image.repository}:latest'
)
ENV_TAGS = (
# gitlab
'CI_COMMIT_SHORT_SHA',
'CI_COMMIT_REF_NAME',
'CI_COMMIT_TAG',
# CircleCI
'CIRCLE_SHA1',
'CIRCLE_TAG',
'CIRCLE_BRANCH',
# contributions welcome here
)
# figure tags from CI vars
for name in ENV_TAGS:
value = os.getenv(name)
if value:
self.image.tags.append(value)
if self.image.tags:
tags = [f'{self.image.repository}:{tag}' for tag in self.image.tags]
else:
tags = [self.image.repository]
for tag in tags:
await self.exec('buildah', 'tag', self.sha, tag, buildah=False)
await self.parent.exec('buildah', 'tag', self.image.repository + ':final', *tags)
async def clean(self, *args, **kwargs):
if self.is_runnable():
for src, dst in self.mounts.items():
await self.exec('umount', self.mnt / str(dst)[1:], buildah=False)
async def mkdir(self, *paths):
return await self.parent.mkdir(*[self.path(path) for path in paths])
if self.status == 'success':
await self.commit()
if 'push' in args:
await self.image.push(action=self)
async def copy(self, *args):
return await self.parent.exec('buildah', 'copy', self.ctr, *args)
if self.mnt is not None:
await self.exec('buildah', 'umount', self.ctr, buildah=False)
async def write(self, path, content):
return await self.write(path, content)
if self.ctr is not None:
await self.exec('buildah', 'rm', self.ctr, buildah=False)
async def write(self, path, content, **kwargs):
return await self.exec(
f'cat > {path} <<EOF\n'
+ content
+ '\nEOF',
**kwargs
)
class Config:
def __init__(self, **config):
self.config = config
async def __call__(self, target):
for key, value in self.config.items():
await target.parent.exec(
f'buildah config --{key} "{value}" {target.ctr}'
)
def __str__(self):
return f'Buildah.Config({self.config})'
class Env:
def __init__(self, **env):
self.env = env
async def __call__(self, target):
for key, value in self.env.items():
await target.parent.exec(
'buildah',
'config',
'--env',
f'{key}={value}',
target.ctr,
)
def __str__(self):
return f'Buildah.Env({self.env})'

76
shlax/targets/docker.py
View File

@ -1,76 +0,0 @@
import asyncio
from pathlib import Path
import os
from ..image import Image
from .localhost import Localhost
class Docker(Localhost):
contextualize = Localhost.contextualize + ['mnt', 'ctr', 'mount']
def __init__(self, *args, **kwargs):
self.image = kwargs.get('image', 'alpine')
if not isinstance(self.image, Image):
self.image = Image(self.image)
super().__init__(*args, **kwargs)
self.context['ctr'] = None
def shargs(self, *args, daemon=False, **kwargs):
if args[0] == 'docker':
return args, kwargs
extra = []
if 'user' in kwargs:
extra += ['--user', kwargs.pop('user')]
args, kwargs = super().shargs(*args, **kwargs)
if self.context['ctr']:
executor = 'exec'
extra = [self.context['ctr']]
return [self.kwargs.get('docker', 'docker'), executor, '-t'] + extra + list(args), kwargs
executor = 'run'
cwd = os.getcwd()
if daemon:
extra += ['-d']
extra = extra + ['-v', f'{cwd}:{cwd}', '-w', f'{cwd}']
return [self.kwargs.get('docker', 'docker'), executor, '-t'] + extra + [str(self.image)] + list(args), kwargs
async def call(self, *args, **kwargs):
name = kwargs.get('name', os.getcwd()).split('/')[-1]
self.context['ctr'] = (
await self.exec(
'docker', 'ps', '-aq', '--filter',
'name=' + name,
raises=False
)
).out.split('\n')[0]
if 'recreate' in args and self.context['ctr']:
await self.exec('docker', 'rm', '-f', self.context['ctr'])
self.context['ctr'] = None
if self.context['ctr']:
self.context['ctr'] = (await self.exec('docker', 'start', name)).out
return await super().call(*args, **kwargs)
async def copy(self, *args):
src = args[:-1]
dst = args[-1]
await self.mkdir(dst)
procs = []
for s in src:
'''
if Path(s).is_dir():
await self.mkdir(s)
args = ['docker', 'copy', self.ctr, s, Path(dst) / s]
else:
args = ['docker', 'copy', self.ctr, s, dst]
'''
args = ['docker', 'cp', s, self.context['ctr'] + ':' + dst]
procs.append(self.exec(*args))
return await asyncio.gather(*procs)

49
shlax/targets/localhost.py
View File

@ -1,14 +1,14 @@
import os
import copy
import re
from shlax.proc import Proc
from ..output import Output
from ..proc import Proc
from ..result import Result, Results
from ..strategies.script import Script
from .base import Target
class Localhost(Script):
root = '/'
class Localhost(Target):
def shargs(self, *args, **kwargs):
user = kwargs.pop('user', None)
args = [str(arg) for arg in args if args is not None]
@ -24,48 +24,17 @@ class Localhost(Script):
elif user:
args = ['sudo', '-u', user] + args
return args, kwargs
if self.parent:
return self.parent.shargs(*args, **kwargs)
else:
return args, kwargs
async def exec(self, *args, **kwargs):
if 'debug' not in kwargs:
kwargs['debug'] = getattr(self, 'call_kwargs', {}).get('debug', False)
kwargs.setdefault('output', self.output)
kwargs['output'] = self.output
args, kwargs = self.shargs(*args, **kwargs)
proc = await Proc(*args, **kwargs)()
if kwargs.get('wait', True):
await proc.wait()
return proc
async def rexec(self, *args, **kwargs):
kwargs['user'] = 'root'
return await self.exec(*args, **kwargs)
async def env(self, name):
return (await self.exec('echo $' + name)).out
async def which(self, *cmd):
"""
Return the first path to the cmd in the container.
If cmd argument is a list then it will try all commands.
"""
proc = await self.exec('type ' + ' '.join(cmd), raises=False)
result = []
for res in proc.out.split('\n'):
match = re.match('([^ ]+) is ([^ ]+)$', res.strip())
if match:
result.append(match.group(1))
return result
async def copy(self, *args):
args = ['cp', '-ra'] + list(args)
return await self.exec(*args)
async def mount(self, *dirs):
pass
async def mkdir(self, *dirs):
return await self.exec(*['mkdir', '-p'] + list(dirs))

24
shlax/targets/ssh.py
View File

@ -1,17 +1,15 @@
import os
from shlax.proc import Proc
from .localhost import Localhost
from .base import Target
class Ssh(Localhost):
root = '/'
def __init__(self, host, *args, **kwargs):
class Ssh(Target):
def __init__(self, *actions, host, user=None):
self.host = host
super().__init__(*args, **kwargs)
self.user = user
super().__init__(*actions)
def shargs(self, *args, **kwargs):
args, kwargs = super().shargs(*args, **kwargs)
return (['ssh', self.host] + list(args)), kwargs
async def exec(self, *args, user=None, **kwargs):
_args = ['ssh', self.host]
if user == 'root':
_args += ['sudo']
_args += [' '.join([str(a) for a in args])]
return await self.parent.exec(*_args, **kwargs)

23
shlax/targets/stub.py Normal file
View File

@ -0,0 +1,23 @@
from ..proc import Proc
from .base import Target
class ProcStub(Proc):
async def __call__(self, wait=True):
return self
async def communicate(self):
self.communicated = True
return self
async def wait(self):
return self
class Stub(Target):
async def exec(self, *args, **kwargs):
proc = await ProcStub(*args, **kwargs)()
if kwargs.get('wait', True):
await proc.wait()
return proc

65
shlaxfile.py
View File

@ -1,55 +1,20 @@
#!/usr/bin/env shlax
from shlax.contrib.gitlab import *
#!/usr/bin/env python
"""
Shlaxfile for shlax itself.
"""
PYTEST = 'py.test -svv tests'
test = Script(
Pip('.[test]'),
Run(PYTEST),
)
build = Buildah(
'quay.io/podman/stable',
Packages('python38', 'buildah', 'unzip', 'findutils', 'python3-yaml', upgrade=False),
Async(
# python3.8 on centos with pip dance ...
Run('''
curl -o setuptools.zip https://files.pythonhosted.org/packages/42/3e/2464120172859e5d103e5500315fb5555b1e908c0dacc73d80d35a9480ca/setuptools-45.1.0.zip
unzip setuptools.zip
mkdir -p /usr/local/lib/python3.8/site-packages/
sh -c "cd setuptools-* && python3.8 setup.py install"
easy_install-3.8 pip
echo python3.8 -m pip > /usr/bin/pip
chmod +x /usr/bin/pip
'''),
Copy('shlax/', 'setup.py', '/app'),
),
Pip('/app[full]'),
commit='docker.io/yourlabs/shlax',
workdir='/app',
)
from shlax.shortcuts import *
shlax = Container(
build=build,
test=Script(Run('./shlaxfile.py -d test')),
build=Buildah(
Packages('python38', 'buildah', 'unzip', 'findutils', upgrade=False),
Copy('setup.py', 'shlax', '/app'),
Pip('/app[cli]'),
base='quay.io/buildah/stable',
commit='docker://docker.io/yourlabs/shlax',
),
)
gitlabci = GitLabCI(
test=dict(
stage='build',
script='pip install -U --user -e .[test] && ' + PYTEST,
image='yourlabs/python',
),
build=dict(
stage='build',
image='yourlabs/shlax',
script='pip install -U --user -e . && CACHE_DIR=$(pwd)/.cache ./shlaxfile.py -d shlax build push',
cache=dict(paths=['.cache'], key='cache'),
),
pypi=dict(
stage='deploy',
only=['tags'],
image='yourlabs/python',
script='pypi-release',
),
)
if __name__ == '__main__':
print(Group(doc=__doc__).load(shlax).entry_point())

53
tests/actions/test_base.py
View File

@ -1,53 +0,0 @@
from shlax import *
inner = Run()
other = Run('ls')
middle = Buildah('alpine', inner, other)
outer = Localhost(middle)
middle = outer.actions[0]
other = middle.actions[1]
inner = middle.actions[0]
def test_action_init_args():
assert other.args == ('ls',)
def test_action_parent_autoset():
assert list(outer.actions) == [middle]
assert middle.parent == outer
assert inner.parent == middle
assert other.parent == middle
def test_action_context():
assert outer.context is inner.context
assert middle.context is inner.context
assert middle.context is outer.context
assert other.context is outer.context
def test_action_sibblings():
assert inner.sibblings() == [other]
assert inner.sibblings(lambda s: s.args[0] == 'ls') == [other]
assert inner.sibblings(lambda s: s.args[0] == 'foo') == []
assert inner.sibblings(type='run') == [other]
assert inner.sibblings(args=('ls',)) == [other]
def test_actions_parents():
assert other.parents() == [middle, outer]
assert other.parents(lambda p: p.base == 'alpine') == [middle]
assert inner.parents(type='localhost') == [outer]
assert inner.parents(type='buildah') == [middle]
def test_action_childrens():
assert middle.children() == [inner, other]
assert middle.children(lambda a: a.args[0] == 'ls') == [other]
assert outer.children() == [middle, inner, other]
def test_action_getattr():
assert other.exec == middle.exec
assert other.shargs == middle.shargs

18
tests/shlaxfail.py Executable file
View File

@ -0,0 +1,18 @@
#!/usr/bin/env python
"""
Shlaxfile for shlax itself.
"""
from shlax.shortcuts import *
shlax = Container(
build=Buildah(
Packages('prout', upgrade=False),
base='alpine',
commit='shlaxfail',
),
)
if __name__ == '__main__':
print(Group(doc=__doc__).load(shlax).entry_point())

17
tests/shlaxsuccess.py Executable file
View File

@ -0,0 +1,17 @@
#!/usr/bin/env python
"""
Shlaxfile for shlax itself.
"""
from shlax.shortcuts import *
shlax = Container(
build=Buildah(
base='alpine',
commit='shlaxsuccess',
),
)
if __name__ == '__main__':
print(Group(doc=__doc__).load(shlax).entry_point())

8
tests/test_image.py
View File

@ -1,7 +1,7 @@
import pytest
import os
from shlax import Image
from shlax.image import Image
tests = {
@ -25,9 +25,3 @@ def test_args(arg, expected):
im = Image(arg)
for k, v in expected.items():
assert getattr(im, k) == v
def test_args_env():
os.environ['IMAGE_TEST_ARGS_ENV'] = 'foo'
Image.ENV_TAGS = ['IMAGE_TEST_ARGS_ENV']
im = Image('re/po:x,y')
assert im.tags == ['x', 'y', 'foo']

2
tests/test_output.py
View File

@ -1,5 +1,5 @@
import pytest
from shlax import Output
from shlax.output import Output
class Write:

65
tests/test_proc.py
View File

@ -1,65 +0,0 @@
import pytest
from unittest.mock import patch
from shlax import *
from shlax import proc
test_args_params = [
(
Localhost(Run('echo hi')),
[('sh', '-euc', 'echo hi')]
),
(
Localhost(Run('echo hi', user='jimi')),
[('sudo', '-u', 'jimi', 'sh', '-euc', 'echo hi')]
),
(
Localhost(Run('echo hi', user='root')),
[('sudo', 'sh', '-euc', 'echo hi')]
),
(
Ssh('host', Run('echo hi', user='root')),
[('ssh', 'host', 'sudo', 'sh', '-euc', 'echo hi')]
),
(
Buildah('alpine', Run('echo hi')),
[
('buildah', 'from', 'alpine'),
('buildah', 'mount', ''),
('buildah', 'run', '', '--', 'sh', '-euc', 'echo hi'),
('buildah', 'umount', ''),
('buildah', 'rm', ''),
]
),
(
Buildah('alpine', Run('echo hi', user='root')),
[
('buildah', 'from', 'alpine'),
('buildah', 'mount', ''),
('buildah', 'run', '--user', 'root', '', '--', 'sh', '-euc', 'echo hi'),
('buildah', 'umount', ''),
('buildah', 'rm', ''),
]
),
(
Ssh('host', Buildah('alpine', Run('echo hi', user='root'))),
[
('ssh', 'host', 'buildah', 'from', 'alpine'),
('ssh', 'host', 'buildah', 'mount', ''),
('ssh', 'host', 'buildah', 'run', '--user', 'root', '', '--', 'sh', '-euc', 'echo hi'),
('ssh', 'host', 'buildah', 'umount', ''),
('ssh', 'host', 'buildah', 'rm', ''),
]
),
]
@pytest.mark.parametrize(
'script,commands',
test_args_params
)
@pytest.mark.asyncio
async def test_args(script, commands):
with Proc.mock():
await script()
assert commands == Proc.test

132
tests/test_target.py Normal file
View File

@ -0,0 +1,132 @@
import pytest
from shlax.targets.stub import Stub
from shlax.actions.run import Run
from shlax.actions.parallel import Parallel
from shlax.result import Result
class Error:
async def __call__(self, target):
raise Exception('lol')
@pytest.mark.asyncio
async def test_success():
action = Run('echo hi')
target = Stub(action)
await target()
assert target.results[0].action == action
assert target.results[0].status == 'success'
@pytest.mark.asyncio
async def test_error():
action = Error()
target = Stub(action)
await target()
assert target.results[0].action == action
assert target.results[0].status == 'failure'
@pytest.mark.asyncio
async def test_nested():
nested = Error()
class Nesting:
async def __call__(self, target):
await target(nested)
nesting = Nesting()
target = Stub(nesting)
await target()
assert len(target.results) == 2
assert target.results[0].status == 'failure'
assert target.results[0].action == nested
assert target.results[1].status == 'failure'
assert target.results[1].action == nesting
@pytest.mark.asyncio
async def test_parallel():
winner = Run('echo hi')
looser = Error()
parallel = Parallel(winner, looser)
target = Stub(parallel)
await target()
assert len(target.results) == 3
assert target.results[0].status == 'success'
assert target.results[0].action == winner
assert target.results[1].status == 'failure'
assert target.results[1].action == looser
assert target.results[2].status == 'failure'
assert target.results[2].action == parallel
@pytest.mark.asyncio
async def test_function():
async def hello(target):
await target.exec('hello')
await Stub()(hello)
@pytest.mark.asyncio
async def test_action_clean():
class Example:
def __init__(self):
self.was_called = False
async def clean(self, target, result):
self.was_called = True
async def __call__(self, target):
raise Exception('lol')
action = Example()
target = Stub()
with pytest.raises(Exception):
await target(action)
assert action.was_called
@pytest.mark.asyncio
async def test_target_clean():
class Example(Stub):
def __init__(self, action):
self.was_called = False
super().__init__(action)
async def clean(self, target, result):
self.was_called = True
target = Example(Error())
await target()
assert target.was_called
@pytest.mark.asyncio
async def test_method():
class Example:
def __init__(self):
self.was_called = False
async def test(self, target):
self.was_called = True
example = Example()
action = example.test
target = Stub()
await target(action)
assert example.was_called
@pytest.mark.asyncio
async def test_target_action():
child = Stub(Run('echo hi'))
parent = Stub(child)
grandpa = Stub()
await grandpa(parent)
assert len(grandpa.results) == 3
grandpa = Stub(parent)
await grandpa()
assert len(grandpa.results) == 3