A => .build.yml +43 -0
@@ 0,0 1,43 @@
+image: alpine/old
+packages:
+ - findutils
+ - linux-headers
+ - libffi-dev
+ - libressl-dev
+ - python3
+ - python3-dev
+ - py3-pip
+ - py3-requests
+# Enable this to debug
+# shell: true
+secrets:
+ - 262ecd1a-0945-48f7-920f-367c59e107bc
+ - 0e73e604-2a62-48ae-b5b6-9b6b0e09fc4d
+ - ea368a24-78eb-4f2c-bca2-52a791042348
+sources:
+ - hg+ssh://hg@hg.sr.ht/~ocurero/pyjed
+tasks:
+ - sync_github: |
+ cd pyjed
+ sudo pip3 -q install hg-git
+ sudo pip3 -q install dulwich
+ echo "[extensions]" >>./.hg/hgrc
+ echo "hgext.bookmarks =" >>./.hg/hgrc
+ echo "hggit = " >>./.hg/hgrc
+ ssh-keyscan -H github.com >> ~/.ssh/known_hosts
+ hg bookmark -r default master # so a ref gets created
+ hg push git+ssh://git@github.com/ocurero/pyjed.git || hg push git+ssh://git@github.com/ocurero/pyjed.git | grep "no changes found"
+ - clone_github: |
+ rm -R pyjed && git clone git@github.com:ocurero/pyjed.git
+ - pass_tests: |
+ cd pyjed
+ curl -sSL https://raw.githubusercontent.com/python-poetry/poetry/master/get-poetry.py | python3
+ source $HOME/.poetry/env
+ poetry install --quiet
+ poetry run tox
+ # - update_readthedocs: |
+ # curl -X POST -d @/home/build/RTD_TOKEN https://readthedocs.org/api/v2/webhook/sqlalchemy-querybuilder/139469/
+ - upload_codecov: |
+ cd pyjed && sudo pip3 -q install coverage
+ curl -s https://codecov.io/bash | bash -s -- -Z -t @/home/build/CODECOV_TOKEN
+triggers: null
A => .github/README.md +38 -0
@@ 0,0 1,38 @@
+### WARNING, this repository is a read-only mirror!
+*See [sourcehut](https://sr.ht/~ocurero/sqlalchemy-querybuilder/) for PR and latest news*
+
+SQLAlchemy query builder for jQuery QueryBuilder
+================================================
+
+[](https://www.repostatus.org/#active) [](https://builds.sr.ht/~ocurero/sqlalchemy-querybuilder/.build.yml?) [](https://codecov.io/gh/ocurero/sqlalchemy-querybuilder) [](https://sqlalchemy-querybuilder.readthedocs.io/)
+
+This package implements a sqlalchemy query builder for json data
+generated with (but not limited to) [`jQuery QueryBuilder`](http://querybuilder.js.org/).
+
+* Open Source: Apache 2.0 license.
+* Website: <https://sr.ht/~ocurero/sqlalchemy-querybuilder/>
+* Documentation: <https://sqlalchemy-querybuilder.readthedocs.io/>
+
+Quickstart
+----------
+
+Using **sqlalchemy-querybuilder** is very simple:
+
+```python
+
+from sqlalchemy_querybuilder import Filter
+from myapp import models, query
+
+ rules = {
+ "condition": "OR",
+ "rules": [{
+ "field": "mytable.myfield",
+ "operator": "equal",
+ "value": "foo"
+ },
+ ],
+ }
+
+ myfilter = Filter(models, query)
+ print(myfilter.querybuilder(rules))
+```
A => .hgignore +45 -0
@@ 0,0 1,45 @@
+syntax: glob
+
+.gitignore
+node_modules
+.c9revisions
+.DS_Store
+Thumbs.db
+
+*.pyc
+*.pyo
+*.pyd
+
+# C extensions
+*.so
+
+# Packages
+*.egg
+*.egg-info
+dist
+build
+eggs
+parts
+bin
+var
+sdist
+develop-eggs
+.installed.cfg
+lib
+lib64
+include
+man
+htmlcov
+__pycache__
+.pytest_cache
+
+# Installer logs
+pip-log.txt
+
+# Unit test / coverage reports
+.coverage
+.tox
+nosetests.xml
+
+# Translations
+*.mo
A => codecov.yml +2 -0
@@ 0,0 1,2 @@
+codecov:
+ branch: master
M pyjed/__init__.py +6 -2
@@ 1,7 1,11 @@
-from pyjed.jcl import (DD,
+from pyjed.jcl import (COMMENT,
+ DD,
EXEC,
JOB,
+ PROC,
+ SET,
+ from_string,
)
__version__ = '0.1.0'
-__all__ = ('DD', 'EXEC', 'JOB')
+__all__ = ('from_string', 'COMMENT', 'DD', 'EXEC', 'JOB', 'PROC', 'SET')
M pyjed/jcl.py +178 -30
@@ 1,10 1,12 @@
-from collections import UserList
+from collections import UserList, UserDict
from dataclasses import dataclass, field, fields, make_dataclass
+import csv
+import shlex
@dataclass
class Base:
- def to_string(self, statement):
+ def _to_string(self, statement):
for parm in fields(self):
if (parm.name != 'name' and
getattr(self, parm.name) is not None and
@@ 16,21 18,48 @@ class Base:
if len(statement[-1]) + len(f'{name}={value},') > 78:
statement.append('// ')
statement[-1] = statement[-1] + f'{name}={value},'
- print(statement)
- statement[-1] = statement[-1][:-1]
+ if statement[-1][-1] in (',', ' '):
+ statement[-1] = statement[-1][:-1]
return statement
@dataclass
+class Base_PROC(UserList, Base):
+ name: str
+ pend: str = field(default='', repr=False)
+ _attrs = ('pend')
+
+ def __post_init__(self):
+ super().__init__()
+
+ def _to_string(self):
+ stream = super()._to_string([f'//{self.name} PROC '])
+ for statement in self.data:
+ stream.extend(statement._to_string())
+ print(stream)
+ return stream + PEND(self.pend)._to_string()
+
+
+class PROC:
+ def __new__(self, name, **kwargs):
+ proc_parms = [(parm, str, field(default=None)) for parm in kwargs]
+ proc_class = make_dataclass(name, proc_parms, bases=(Base_PROC,))
+ return proc_class(name, **kwargs)
+
+
+@dataclass
class DISP:
status: str = ''
normal: str = ''
abnormal: str = ''
+class DD_Concat(UserList):
+ pass
+
+
@dataclass
class DD(Base):
- name: str
stream: str = field(default=None, repr=False)
accode: str = field(default=None, repr=False)
amp: str = field(default=None, repr=False)
@@ 105,34 134,46 @@ class DD(Base):
_attrs = ('stream')
_alias = {'VOLUME': 'VOL'}
+ _unserialize_attrs = ()
- def to_string(self):
+ def _to_string(self, statements):
if self.stream:
- stream = super().to_string([f'//{self.name} DD * '])
+ stream = super()._to_string([statements[-1] + '*'])
stream.append(self.stream)
return stream
else:
- return super().to_string([f'//{self.name} DD '])
+ return super()._to_string(statements)
+
+ @classmethod
+ def _from_string(cls, exec, dd_parms):
+ dd_name = dd_parms[0][2:]
+ dd = cls()
+ for name, value in _parms_as_dict(dd_parms[2:]).items():
+ name = name if name not in dd._unserialize_attrs else '_' + name
+ setattr(dd, name, value)
+ if dd_name:
+ exec[dd_name] = [dd]
+ else:
+ dd_name = list(exec.keys())[-1]
+ exec[dd_name].append(dd)
+ return exec
@dataclass
-class COMMENT:
- text: str = ''
-
-
-@dataclass
-class STEP(Base):
+class STEP(UserDict, Base):
name: str
- dd: list = field(default_factory=list)
- _attrs = ('dd')
+ _attrs = ()
- def to_string(self):
- string = super().to_string([f'//{self.name} EXEC '])
- for dd in self.dd:
- string = string + dd.to_string()
- print(string)
- print('KIKO IKIKO IKIKO IKOKI')
- return string
+ def _to_string(self):
+ statements = super()._to_string([f'//{self.name} EXEC '])
+ for ddname, ddlist in self.data.items():
+ for dd in ddlist:
+ statements.extend(dd._to_string([f'//{ddname} DD ']))
+ ddname = ''
+ return statements
+
+ def __post_init__(self):
+ super().__init__()
class EXEC():
@@ 154,7 195,7 @@ class EXEC():
region: str = field(default=None, repr=False)
rlstmout: str = field(default=None, repr=False)
time: str = field(default=None, repr=False)
- steplib: DD = field(default=None)
+ _unserialize_attrs = {}
def __new__(self, *args, **kwargs):
if len(args) == 2 or 'procname' in kwargs:
@@ 165,6 206,61 @@ class EXEC():
else:
return EXEC.EXEC(*args, **kwargs)
+ @classmethod
+ def _from_string(cls, job, statements):
+ print(statements)
+ exec_parms = statements.pop()
+ exec = cls(exec_parms[0][2:])
+ for name, value in _parms_as_dict(exec_parms[2:]).items():
+ name = name if name not in exec._unserialize_attrs else '_' + name
+ setattr(exec, name, value)
+ job.append(exec)
+ job, statements = COMMENT._from_string(job, statements)
+ while statements and statements[-1][1] == 'DD':
+ exec = DD._from_string(exec, statements[-1])
+ statements.pop()
+ return job, statements
+
+
+@dataclass
+class PEND(Base):
+ name: str = field(repr=False)
+ _attrs = ()
+
+ def _to_string(self):
+ return super()._to_string([f'//{self.name} PEND'])
+
+
+@dataclass
+class Base_SET(Base):
+ name: str = field(default=None, repr=False)
+ _attrs = ()
+
+ def _to_string(self):
+ return super()._to_string([f'//{self.name} SET '])
+
+
+class SET:
+ def __new__(self, name='', **kwargs):
+ proc_parms = [(parm, str, field(default=None)) for parm in kwargs]
+ proc_class = make_dataclass(name, proc_parms, bases=(Base_SET,))
+ return proc_class(name, **kwargs)
+
+
+@dataclass()
+class COMMENT(str, Base):
+
+ @classmethod
+ def _from_string(cls, job, statements):
+ comment = cls()
+ while (statements and isinstance(statements[-1][0], str)
+ and statements[-1][0][:3].startswith('//*')):
+ comment += statements.pop()[0][3:] + '\n'
+
+ if comment:
+ job.append(comment[:-1])
+ return job, statements
+
@dataclass()
class JOB(UserList, Base):
@@ 198,13 294,65 @@ class JOB(UserList, Base):
addrspc: str = field(default=None, repr=False)
_attrs = ('steps')
+ _unserialize_attrs = {'class', '_class'}
+
+ def to_string(self):
+ stream = super()._to_string([f'//{self.name} JOB '])
+ for statement in self.data:
+ stream.extend(statement._to_string())
+ return '\n'.join(stream)
def __post_init__(self):
super().__init__()
- def to_string(self):
- stream = super().to_string([f'//{self.name} JOB '])
- for statement in self.data:
- stream.extend(statement.to_string())
- print(stream)
- return '\n'.join(stream)
+ @classmethod
+ def _from_string(cls, statements):
+ statements.reverse()
+ job_parms = statements.pop()
+ job = cls(job_parms[0][2:])
+ for name, value in _parms_as_dict(job_parms[2:]).items():
+ name = name if name not in job._unserialize_attrs else '_' + name
+ setattr(job, name, value)
+ while statements:
+ if statements[-1][1] == 'EXEC':
+ job, statements = EXEC._from_string(job, statements)
+ return job
+
+
+def from_string(stream):
+ statements = []
+ statement = []
+ for line in stream.splitlines():
+ tokens = []
+ if line[-1] == ',':
+ line = line + '\t'
+ if line[:2] == '//' and line[:3] != '//*':
+ for token in csv.reader(shlex.split(line), skipinitialspace=True):
+ for parm in token:
+ if parm:
+ tokens.append(parm)
+ if line[-1] == '\t':
+ if not statement:
+ statement = tokens
+ else:
+ statement.extend(tokens[1:])
+ continue
+ if statement:
+ tokens = tokens[1:]
+ statement.extend(tokens)
+ else:
+ statement = [line]
+ statements.extend(statement)
+ statement = []
+ else:
+ statements.extend(statement)
+
+ return JOB._from_string(statements)
+
+
+def _parms_as_dict(parm_list):
+ parm_dict = {}
+ for parm in parm_list:
+ name, value = parm.split('=')
+ parm_dict[name.lower()] = value
+ return parm_dict
M pyproject.toml +3 -0
@@ 10,6 10,9 @@ dataclasses = { version="^0.8", python =
[tool.poetry.dev-dependencies]
pytest = "^5.2"
+pytest-mock = "^3.6.1"
+coverage = "^6.2"
+tox = "^3.24.5"
[build-system]
requires = ["poetry-core>=1.0.0"]
M tests/test_pyjed.py => tests/test_serialize.py +85 -32
@@ 1,16 1,12 @@
-# import pytest
+# import pytest
from pyjed import __version__
-from pyjed import DD, EXEC, JOB
+from pyjed import DD, EXEC, JOB, PROC, SET
def test_version():
assert __version__ == '0.1.0'
-def test_job():
- JOB('MYJOB', _class='S')
-
-
def test_serialize_job_one_line():
job = JOB('MYJOB', _class='S')
assert job.to_string() == '//MYJOB JOB CLASS=S'
@@ 30,69 26,126 @@ def test_serialize_job_with_step():
assert job.to_string() == '//MYJOB JOB CLASS=S\n//STEP1 EXEC PGM=IEFBR14'
-def test_exec():
+def test_serialize_exec_proc():
+ proc = EXEC('STEP', procname='PROC1', test=1, foo=2)
+ proc._to_string()
+
+
+def test_serialize_dd_dsname():
+ DD(dsname='MY.DATASET')
+
+
+def test_serialize_dd_dataset_no_concat_one_line():
+ dd = DD(dsname='MY.DATASET')
+ assert dd._to_string(['//MYDD DD ']) == ['//MYDD DD DSNAME=MY.DATASET']
+
+
+def test_serialize_dd_instream():
+ dd = DD(stream='Lorem ipsum dolor sit amet')
+ assert dd._to_string(['//MYDD DD ']) == ['//MYDD DD *', 'Lorem ipsum dolor sit amet']
+
+
+def test_serialize_dd_dataset_two_lines():
+ dd = DD(dsname='MY.LONG.LONG.LONG.DATASET', volume='SER',
+ disp='(,CATLG,DELETE)', dataclas='DCLAS02')
+ assert dd._to_string(['//MYDD DD ']) == [
+ '//MYDD DD DATACLAS=DCLAS02,DISP=(,CATLG,DELETE),',
+ '// DSNAME=MY.LONG.LONG.LONG.DATASET,VOLUME=SER']
+
+
+def test_serialize_dd_unixpath_one_line():
+ dd = DD(path='/usr/applics/pay.time')
+ assert dd._to_string(['//MYDD DD ']) == ['//MYDD DD PATH=\'/usr/applics/pay.time\'']
+
+
+def test_serialize_exec():
EXEC('STEP1')
-def test_exec_proc_arg():
+def test_serialize_exec_proc_arg():
proc = EXEC('STEP', 'PROC1', test=1, foo=2)
assert proc.test == 1 and proc.foo == 2
-def test_exec_proc_kwarg():
+def test_serialize_exec_proc_kwarg():
proc = EXEC('STEP', procname='PROC1', test=1, foo=2)
assert proc.test == 1 and proc.foo == 2
def test_serialize_exec_one_line():
exec = EXEC('STEP1', pgm='IEFBR14')
- assert exec.to_string() == ['//STEP1 EXEC PGM=IEFBR14']
+ assert exec._to_string() == ['//STEP1 EXEC PGM=IEFBR14']
def test_serialize_exec_one_line_with_whitespace_parm():
exec = EXEC('STEP1', pgm='IEFBR14', parm='HELLO WORLD')
- assert exec.to_string() == ['//STEP1 EXEC PARM=\'HELLO WORLD\',PGM=IEFBR14']
+ assert exec._to_string() == ['//STEP1 EXEC PARM=\'HELLO WORLD\',PGM=IEFBR14']
def test_serialize_exec_one_line_with_lowercase_parm():
exec = EXEC('STEP1', pgm='IEFBR14', parm='hello world')
- assert exec.to_string() == ['//STEP1 EXEC PARM=\'hello world\',PGM=IEFBR14']
+ assert exec._to_string() == ['//STEP1 EXEC PARM=\'hello world\',PGM=IEFBR14']
def test_serialize_exec_two_lines():
exec = EXEC('STEP1', pgm='IEFBR14', parm='Lorem ipsum dolor sit amet, consectetur '
'adipiscing elit')
- assert exec.to_string() == ['//STEP1 EXEC PARM=\'Lorem ipsum dolor sit amet, '
+ assert exec._to_string() == ['//STEP1 EXEC PARM=\'Lorem ipsum dolor sit amet, '
'consectetur adipiscing elit\',',
'// PGM=IEFBR14']
-def test_serialize_exec_proc():
- proc = EXEC('STEP', procname='PROC1', test=1, foo=2)
- proc.to_string()
+def test_serialize_exec_with_dd():
+ exec = EXEC('STEP1', pgm='IEFBR14')
+ exec['MYDD'] = [DD(dsname='MY.DATASET')]
+ assert exec._to_string() == ['//STEP1 EXEC PGM=IEFBR14',
+ '//MYDD DD DSNAME=MY.DATASET']
-def test_dd_dsname():
- DD('MYDD', dsname='MY.DATASET')
+def test_serialize_exec_with_dd_concat():
+ exec = EXEC('STEP1', pgm='IEFBR14')
+ exec['MYDD'] = [DD(dsname='MY.DATASET'), DD(dsname='MY.DATASET2')]
+ assert exec._to_string() == ['//STEP1 EXEC PGM=IEFBR14',
+ '//MYDD DD DSNAME=MY.DATASET',
+ '// DD DSNAME=MY.DATASET2']
-def test_serialize_dd_dataset_one_line():
- dd = DD('MYDD', dsname='MY.DATASET')
- assert dd.to_string() == ['//MYDD DD DSNAME=MY.DATASET']
+def test_serialize_proc_without_parms():
+ proc = PROC('MYPROC')
+ assert proc._to_string() == ['//MYPROC PROC',
+ '// PEND']
+
+
+def test_serialize_proc_with_parms():
+ proc = PROC('MYPROC', foo=1, bar=2)
+ assert proc._to_string() == ['//MYPROC PROC FOO=1,BAR=2',
+ '// PEND']
-def test_serialize_dd_instream():
- dd = DD('MYDD', stream='Lorem ipsum dolor sit amet')
- assert dd.to_string() == ['//MYDD DD *', 'Lorem ipsum dolor sit amet']
+def test_serialize_proc_with_exec():
+ proc = PROC('MYPROC', foo=1, bar=2)
+ proc.append(EXEC('STEP1', pgm='IEFBR14'))
+ assert proc._to_string() == ['//MYPROC PROC FOO=1,BAR=2',
+ '//STEP1 EXEC PGM=IEFBR14',
+ '// PEND']
-def test_serialize_dd_dataset_two_lines():
- dd = DD('MYDD', dsname='MY.LONG.LONG.LONG.DATASET', volume='SER',
- disp='(,CATLG,DELETE)', dataclas='DCLAS02')
- assert dd.to_string() == ['//MYDD DD DATACLAS=DCLAS02,DISP=(,CATLG,DELETE),',
- '// DSNAME=MY.LONG.LONG.LONG.DATASET,VOLUME=SER']
+def test_serialize_proc_with_exec_and_dd():
+ proc = PROC('MYPROC', foo=1, bar=2)
+ exec = EXEC('STEP1', pgm='IEFBR14')
+ exec['MYDD'] = [DD(dsname='MY.DATASET')]
+ proc.append(exec)
+ assert proc._to_string() == ['//MYPROC PROC FOO=1,BAR=2',
+ '//STEP1 EXEC PGM=IEFBR14',
+ '//MYDD DD DSNAME=MY.DATASET',
+ '// PEND']
-def test_serialize_dd_unixpath_one_line():
- dd = DD('MYDD', path='/usr/applics/pay.time')
- assert dd.to_string() == ['//MYDD DD PATH=\'/usr/applics/pay.time\'']
+def test_serialize_set_with_name():
+ set = SET('MYSET', foo=1, bar=2)
+ assert set._to_string() == ['//MYSET SET FOO=1,BAR=2']
+
+
+def test_serialize_set_without_name():
+ set = SET(foo=1, bar=2)
+ assert set._to_string() == ['// SET FOO=1,BAR=2']
A => tests/test_unserialize.py +177 -0
@@ 0,0 1,177 @@
+# import pytest
+from pyjed import __version__
+from pyjed import from_string, COMMENT, DD, EXEC, JOB, PROC, SET
+
+
+def test_version():
+ assert __version__ == '0.1.0'
+
+
+def test_from_string(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ job = from_string('//MYJOB JOB CLASS=S')
+ assert hasattr(job, 'name')
+
+
+def test_from_string_continuation_line(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ from_string('//MYJOB JOB CLASS=S,\n// MSGCLASS=V')
+ JOB._from_string.assert_called_once_with(['//MYJOB',
+ 'JOB',
+ 'CLASS=S',
+ 'MSGCLASS=V'])
+
+
+def test_from_string_continuation_line_two_lines(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ from_string('//MYJOB JOB CLASS=S,\n// MSGCLASS=V,\n// NOTIFY=&SYSUID')
+ JOB._from_string.assert_called_once_with(['//MYJOB',
+ 'JOB',
+ 'CLASS=S',
+ 'MSGCLASS=V',
+ 'NOTIFY=&SYSUID'])
+
+
+def test_from_string_continuation_line_with_spaces(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ from_string('//MYJOB JOB \'JOB PRO\',CLASS=S,\n// MSGCLASS=V')
+ JOB._from_string.assert_called_once_with(['//MYJOB',
+ 'JOB',
+ 'JOB PRO',
+ 'CLASS=S',
+ 'MSGCLASS=V'])
+
+
+def test_from_string_instream_DD(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ from_string('//MYJOB JOB CLASS=S\n//MYDD DD *\nLorem ipsum\ndolor sit amet')
+ JOB._from_string.assert_called_once_with(['//MYJOB',
+ 'JOB',
+ 'CLASS=S',
+ '//MYDD',
+ 'DD',
+ '*',
+ 'Lorem ipsum',
+ 'dolor sit amet'])
+
+
+def test_from_string_comment_one_line(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ from_string('//MYJOB JOB CLASS=S\n//*LOREM IPSUM')
+ JOB._from_string.assert_called_once_with(['//MYJOB',
+ 'JOB',
+ 'CLASS=S',
+ '//*LOREM IPSUM'])
+
+
+def test_from_string_comment_two_lines(mocker):
+ mocker.patch('pyjed.JOB._from_string')
+ from_string('//MYJOB JOB CLASS=S\n//*LOREM IPSUM\n//*DOLOR SIT AMET')
+ JOB._from_string.assert_called_once_with(['//MYJOB',
+ 'JOB',
+ 'CLASS=S',
+ '//*LOREM IPSUM',
+ '//*DOLOR SIT AMET'])
+
+
+def test_unserialize_job_name():
+ job = JOB._from_string([['//MYJOB', 'JOB', 'CLASS=S']])
+ assert job.name == 'MYJOB'
+
+
+def test_unserialize_job_parameters():
+ job = JOB._from_string([['//MYJOB', 'JOB', 'CLASS=S']])
+ assert job._class == 'S'
+
+
+def test_unserialize_job_with_exec():
+ job = JOB._from_string([['//MYJOB', 'JOB', 'CLASS=S'],
+ ['//STEP1', 'EXEC', 'PGM=IEFBR14']])
+ assert job[0].name == 'STEP1'
+
+
+def test_unserialize_comment_one_line():
+ job, _ = COMMENT._from_string(JOB(''), [['//*LOREM IPSUM DOLOR SIT AMET']])
+ assert job[0] == 'LOREM IPSUM DOLOR SIT AMET'
+
+
+def test_unserialize_comment_two_lines():
+ job, _ = COMMENT._from_string(JOB(''), [['//*DOLOR SIT AMET'],
+ ['//*LOREM IPSUM']])
+ assert job[0] == 'LOREM IPSUM\nDOLOR SIT AMET'
+
+
+def test_unserialize_exec_with_DD():
+ job, _ = EXEC._from_string(JOB(''), [['//MYDD', 'DD', 'DSNAME=MY.DSN'],
+ ['//STEP1', 'EXEC', 'PGM=IEFBR14']])
+
+ assert 'MYDD' in job[0] and job[0]['MYDD'][0].dsname == 'MY.DSN'
+
+
+def test_unserialize_exec_with_concat_DD():
+ job, _ = EXEC._from_string(JOB('MYJOB'), [['//', 'DD', 'DSNAME=MY.DSN2'],
+ ['//MYDD', 'DD', 'DSNAME=MY.DSN'],
+ ['//STEP1', 'EXEC', 'PGM=IEFBR14']])
+
+ assert 'MYDD' in job[0] and job[0]['MYDD'][1].dsname == 'MY.DSN2'
+
+
+def tst_unserialize_exec_proc():
+ proc = EXEC('STEP', procname='PROC1', test=1, foo=2)
+ proc._to_string()
+
+
+def tst_unserialize_dd_dsname():
+ DD(dsname='MY.DATASET')
+
+
+def tst_unserialize_dd_dataset_no_concat_one_line():
+ dd = DD(dsname='MY.DATASET')
+ assert dd._to_string(['//MYDD DD ']) == ['//MYDD DD DSNAME=MY.DATASET']
+
+
+def tst_unserialize_dd_instream():
+ dd = DD(stream='Lorem ipsum dolor sit amet')
+ assert dd._to_string(['//MYDD DD ']) == ['//MYDD DD *', 'Lorem ipsum dolor sit amet']
+
+
+def tst_unserialize_dd_dataset_two_lines():
+ dd = DD(dsname='MY.LONG.LONG.LONG.DATASET', volume='SER',
+ disp='(,CATLG,DELETE)', dataclas='DCLAS02')
+ assert dd._to_string(['//MYDD DD ']) == [
+ '//MYDD DD DATACLAS=DCLAS02,DISP=(,CATLG,DELETE),',
+ '// DSNAME=MY.LONG.LONG.LONG.DATASET,VOLUME=SER']
+
+
+def tst_unserialize_dd_unixpath_one_line():
+ dd = DD(path='/usr/applics/pay.time')
+ assert dd._to_string(['//MYDD DD ']) == ['//MYDD DD PATH=\'/usr/applics/pay.time\'']
+
+
+def tst_unserialize_exec():
+ EXEC('STEP1')
+
+
+def tst_unserialize_exec_proc_arg():
+ proc = EXEC('STEP', 'PROC1', test=1, foo=2)
+ assert proc.test == 1 and proc.foo == 2
+
+
+def tst_unserialize_exec_proc_kwarg():
+ proc = EXEC('STEP', procname='PROC1', test=1, foo=2)
+ assert proc.test == 1 and proc.foo == 2
+
+
+def tst_unserialize_exec_one_line():
+ exec = EXEC('STEP1', pgm='IEFBR14')
+ assert exec._to_string() == ['//STEP1 EXEC PGM=IEFBR14']
+
+
+def tst_unserialize_exec_one_line_with_whitespace_parm():
+ exec = SET('STEP1', pgm='IEFBR14', parm='HELLO WORLD')
+ assert exec._to_string() == ['//STEP1 EXEC PARM=\'HELLO WORLD\',PGM=IEFBR14']
+
+
+def tst_unserialize_exec_one_line_with_lowercase_parm():
+ exec = PROC('STEP1', pgm='IEFBR14', parm='hello world')
+ assert exec._to_string() == ['//STEP1 EXEC PARM=\'hello world\',PGM=IEFBR14']
A => tox.ini +19 -0
@@ 0,0 1,19 @@
+# tox (https://tox.readthedocs.io/) is a tool for running tests
+# in multiple virtualenvs. This configuration file will run the
+# test suite on all supported python versions. To use it, "pip install tox"
+# and then run "tox" from this directory.
+
+[tox]
+# envlist = sqlalchemy14,sqlalchemy13
+envlist = pyjed
+isolated_build = True
+
+[testenv]
+deps =
+ pytest
+ pytest-mock
+ coverage
+ pyjed
+
+commands =
+ coverage run --source=pyjed -m pytest