wrynegade/scwrypts
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

hokay first iteration of python-dudes is ready

Browse Source
This commit is contained in:
Wryn (yage) Wagner 2024-02-19 18:25:34 -07:00
parent fb8bfa6a9c
commit cdb30f2dc0
34 changed files with 1130 additions and 383 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
py/lib/scwrypts/__init__.py
View File

@ -4,6 +4,10 @@ scwrypts
python library functions and invoker for scwrypts
'''
from .scwrypts.execute import execute
from .scwrypts.interactive import interactive
from .scwrypts.scwrypts import scwrypts
__all__ = [
'scwrypts',
'execute',
'interactive',
]
from .scwrypts import scwrypts, execute, interactive

4
py/lib/scwrypts/data/test_converter.py
View File

@ -1,6 +1,4 @@
from io import StringIO
#from string import ascii_letters, digits
from unittest.mock import patch
from pytest import raises
@ -56,7 +54,7 @@ def test_convert_to_yaml():
def test_convert_deep_json_to_yaml():
input_stream = StringIO(generate('json', {**GENERATE_OPTIONS, 'depth': 4}))
input_stream = generate('json', {**GENERATE_OPTIONS, 'depth': 4})
convert(input_stream, 'json', StringIO(), 'yaml')
def test_convert_deep_yaml_to_json():

37
py/lib/scwrypts/http/client.py
View File

@ -1,20 +1,25 @@
from requests import request
def get_request_client(base_url, headers=None):
if headers is None:
headers = {}
CLIENTS = {}
return lambda method, endpoint, **kwargs: request(
method = method,
url = f'{base_url}/{endpoint}',
headers = {
**headers,
**kwargs.get('headers', {}),
},
**{
key: value
for key, value in kwargs.items()
if key != 'headers'
},
)
def get_request_client(base_url, headers=None):
if CLIENTS.get(base_url, None) is None:
if headers is None:
headers = {}
CLIENTS[base_url] = lambda method, endpoint, **kwargs: request(
method = method,
url = f'{base_url}/{endpoint}',
headers = {
**headers,
**kwargs.get('headers', {}),
},
**{
key: value
for key, value in kwargs.items()
if key != 'headers'
},
)
return CLIENTS[base_url]

43
py/lib/scwrypts/http/conftest.py Normal file
View File

@ -0,0 +1,43 @@
from types import SimpleNamespace
from pytest import fixture
from scwrypts.test import generate
from scwrypts.test.character_set import uri
options = {
'str_length_minimum': 8,
'str_length_maximum': 128,
'uuid_output_type': str,
}
def get_request_client_sample_data():
return {
'base_url' : generate(str, options | {'character_set': uri}),
'endpoint' : generate(str, options | {'character_set': uri}),
'method' : generate(str, options),
'response' : generate('requests_Response', options | {'depth': 4}),
'payload' : generate(dict, {
**options,
'depth': 1,
'data_types': { str, 'uuid' },
}),
}
@fixture(name='sample')
def fixture_sample():
return SimpleNamespace(
**get_request_client_sample_data(),
headers = generate(dict, {
**options,
'depth': 1,
'data_types': { str, 'uuid' },
}),
payload_headers = generate(dict, {
**options,
'depth': 1,
'data_types': { str, 'uuid' },
}),
)

20
py/lib/scwrypts/http/directus/__init__.py
View File

@ -1,4 +1,22 @@
from .client import *
'''
basic scwrypts.http client for directus
configured by setting DIRECTUS__BASE_URL and DIRECTUS__API_TOKEN in
scwrypts environment
'''
__all__ = [
'request',
'graphql',
'get_collections',
'get_fields',
'FILTER_OPERATORS',
]
from .client import request
from .graphql import graphql
from .collections import get_collections
from .fields import get_fields
FILTER_OPERATORS = {
'_eq',

57
py/lib/scwrypts/http/directus/client.py
View File

@ -3,55 +3,10 @@ from scwrypts.env import getenv
from .. import get_request_client
REQUEST = None
COLLECTIONS = None
FIELDS = {}
def request(method, endpoint, **kwargs):
global REQUEST # pylint: disable=global-statement
if REQUEST is None:
REQUEST = get_request_client(
base_url = getenv("DIRECTUS__BASE_URL"),
headers = {
'Authorization': f'bearer {getenv("DIRECTUS__API_TOKEN")}',
}
)
return REQUEST(method, endpoint, **kwargs)
def graphql(query, system=False):
return request(
'POST',
'graphql' if system is True else 'graphql/system',
json={'query': query},
)
def get_collections():
global COLLECTIONS # pylint: disable=global-statement
if COLLECTIONS is None:
COLLECTIONS = [
item['collection']
for item in request(
'GET',
'collections?limit=-1&fields[]=collection',
).json()['data']
]
return COLLECTIONS
def get_fields(collection):
if FIELDS.get(collection) is None:
FIELDS[collection] = [
item['field']
for item in request(
'GET',
f'fields/{collection}?limit=-1&fields[]=field',
).json()['data']
]
return FIELDS[collection]
return get_request_client(
base_url = getenv("DIRECTUS__BASE_URL"),
headers = {
'Authorization': f'bearer {getenv("DIRECTUS__API_TOKEN")}',
}
)(method, endpoint, **kwargs)

18
py/lib/scwrypts/http/directus/collections.py Normal file
View File

@ -0,0 +1,18 @@
from .client import request
COLLECTIONS = None
def get_collections():
global COLLECTIONS # pylint: disable=global-statement
if COLLECTIONS is None:
COLLECTIONS = [
item['collection']
for item in request(
'GET',
'collections?limit=-1&fields[]=collection',
).json()['data']
]
return COLLECTIONS

16
py/lib/scwrypts/http/directus/conftest.py Normal file
View File

@ -0,0 +1,16 @@
from types import SimpleNamespace
from pytest import fixture
from scwrypts.test import generate
from scwrypts.test.character_set import uri
from ..conftest import options, get_request_client_sample_data
@fixture(name='sample')
def fixture_sample():
return SimpleNamespace(
**get_request_client_sample_data(),
api_token = generate(str, options | {'character_set': uri}),
query = generate(str, options),
)

16
py/lib/scwrypts/http/directus/fields.py Normal file
View File

@ -0,0 +1,16 @@
from .client import request
FIELDS = {}
def get_fields(collection):
if FIELDS.get(collection) is None:
FIELDS[collection] = [
item['field']
for item in request(
'GET',
f'fields/{collection}?limit=-1&fields[]=field',
).json()['data']
]
return FIELDS[collection]

9
py/lib/scwrypts/http/directus/graphql.py Normal file
View File

@ -0,0 +1,9 @@
from .client import request
def graphql(query, system=False):
return request(
'POST',
'graphql' if system is False else 'graphql/system',
json={'query': query},
)

43
py/lib/scwrypts/http/directus/test_client.py Normal file
View File

@ -0,0 +1,43 @@
from unittest.mock import patch
from pytest import fixture
from .client import request
def test_directus_request(sample, _response):
assert _response == sample.response
def test_directus_request_client_setup(sample, _response, mock_get_request_client):
mock_get_request_client.assert_called_once_with(
base_url = sample.base_url,
headers = { 'Authorization': f'bearer {sample.api_token}' },
)
#####################################################################
@fixture(name='_response')
def fixture_response(sample):
return request(
method = sample.method,
endpoint = sample.endpoint,
**sample.payload,
)
#####################################################################
@fixture(name='mock_getenv', autouse=True)
def fixture_mock_getenv(sample):
with patch('scwrypts.http.directus.client.getenv',) as mock:
mock.side_effect = lambda name: {
'DIRECTUS__BASE_URL': sample.base_url,
'DIRECTUS__API_TOKEN': sample.api_token,
}[name]
yield mock
@fixture(name='mock_get_request_client', autouse=True)
def fixture_mock_get_request_client(sample):
with patch('scwrypts.http.directus.client.get_request_client') as mock:
mock.return_value = lambda method, endpoint, **kwargs: sample.response
yield mock

45
py/lib/scwrypts/http/directus/test_graphql.py Normal file
View File

@ -0,0 +1,45 @@
from unittest.mock import patch
from pytest import fixture
from .graphql import graphql
def test_directus_graphql(sample, _response, _mock_request):
assert _response == sample.response
def test_directus_graphql_request_payload(sample, _response, _mock_request):
_mock_request.assert_called_once_with(
'POST',
'graphql',
json = {'query': sample.query},
)
def test_directus_graphql_system(sample, _response_system):
assert _response_system == sample.response
def test_directus_graphql_system_request_payload(sample, _response_system, _mock_request):
_mock_request.assert_called_once_with(
'POST',
'graphql/system',
json = {'query': sample.query},
)
#####################################################################
@fixture(name='_response')
def fixture_response(sample, _mock_request):
return graphql(sample.query)
@fixture(name='_response_system')
def fixture_response_system(sample, _mock_request):
return graphql(sample.query, system=True)
#####################################################################
@fixture(name='_mock_request')
def fixture_mock_request(sample):
with patch('scwrypts.http.directus.graphql.request') as mock:
mock.return_value = sample.response
yield mock

16
py/lib/scwrypts/http/discord/__init__.py
View File

@ -1,2 +1,14 @@
from .client import *
from .send_message import *
'''
basic scwrypts.http client for discord
configured by setting various DISCORD__* options in the
scwrypts environment
'''
__all__ = [
'request',
'send_message',
]
from .client import request
from .send_message import send_message

23
py/lib/scwrypts/http/discord/client.py
View File

@ -1,20 +1,15 @@
from scwrypts.env import getenv
from scwrypts.http import get_request_client
REQUEST = None
from .. import get_request_client
def request(method, endpoint, **kwargs):
global REQUEST # pylint: disable=global-statement
headers = {}
if REQUEST is None:
headers = {}
if (token := getenv("DISCORD__BOT_TOKEN", required = False)) is not None:
headers['Authorization'] = f'Bot {token}'
if (token := getenv("DISCORD__BOT_TOKEN", required = False)) is not None:
headers['Authorization'] = f'Bot {token}'
REQUEST = get_request_client(
base_url = 'https://discord.com/api',
headers = headers,
)
return REQUEST(method, endpoint, **kwargs)
return get_request_client(
base_url = 'https://discord.com/api',
headers = headers,
)(method, endpoint, **kwargs)

25
py/lib/scwrypts/http/discord/conftest.py Normal file
View File

@ -0,0 +1,25 @@
from string import ascii_letters, digits
from types import SimpleNamespace
from pytest import fixture
from scwrypts.test import generate
from scwrypts.test.character_set import uri
from ..conftest import options, get_request_client_sample_data
@fixture(name='sample')
def fixture_sample():
return SimpleNamespace(
**{
**get_request_client_sample_data(),
'base_url': 'https://discord.com/api',
},
bot_token = generate(str, options | {'character_set': uri}),
username = generate(str, options | {'character_set': ascii_letters + digits}),
avatar_url = generate(str, options | {'character_set': uri}),
webhook = generate(str, options | {'character_set': uri}),
channel_id = generate(str, options | {'character_set': uri}),
content_header = generate(str, options),
content_footer = generate(str, options),
content = generate(str, options),
)

54
py/lib/scwrypts/http/discord/test_client.py Normal file
View File

@ -0,0 +1,54 @@
from unittest.mock import patch
from pytest import fixture
from .client import request
def test_discord_request(sample, _response):
assert _response == sample.response
def test_discord_request_client_setup(sample, mock_get_request_client, _mock_getenv, _response):
mock_get_request_client.assert_called_once_with(
base_url = sample.base_url,
headers = { 'Authorization': f'Bot {sample.bot_token}' },
)
def test_discord_request_client_setup_public(sample, mock_get_request_client, _mock_getenv_optional, _response):
mock_get_request_client.assert_called_once_with(
base_url = sample.base_url,
headers = {},
)
#####################################################################
@fixture(name='_response')
def fixture_response(sample):
return request(
method = sample.method,
endpoint = sample.endpoint,
**sample.payload,
)
#####################################################################
@fixture(name='mock_get_request_client', autouse=True)
def fixture_mock_get_request_client(sample):
with patch('scwrypts.http.discord.client.get_request_client') as mock:
mock.return_value = lambda method, endpoint, **kwargs: sample.response
yield mock
@fixture(name='_mock_getenv')
def fixture_mock_getenv(sample):
with patch('scwrypts.http.discord.client.getenv',) as mock:
mock.side_effect = lambda name, **kwargs: {
'DISCORD__BOT_TOKEN': sample.bot_token,
}[name]
yield mock
@fixture(name='_mock_getenv_optional')
def fixture_mock_getenv_optional():
with patch('scwrypts.http.discord.client.getenv',) as mock:
mock.side_effect = lambda name, **kwargs: None
yield mock

91
py/lib/scwrypts/http/discord/test_send_message.py Normal file
View File

@ -0,0 +1,91 @@
from unittest.mock import patch
from pytest import fixture, raises
from .send_message import send_message
def test_discord_send_message(sample, mock_request, _mock_getenv):
expected = get_default_called_with(sample)
assert send_message(sample.content) == sample.response
mock_request.assert_called_once_with(**expected)
def test_discord_send_message_without_username(sample, mock_request, _mock_getenv):
sample.username = None
expected = get_default_called_with(sample)
del expected['json']['username']
assert send_message(sample.content) == sample.response
mock_request.assert_called_once_with(**expected)
def test_discord_send_message_without_avatar_url(sample, mock_request, _mock_getenv):
sample.avatar_url = None
expected = get_default_called_with(sample)
del expected['json']['avatar_url']
assert send_message(sample.content) == sample.response
mock_request.assert_called_once_with(**expected)
def test_discord_send_message_to_channel_id(sample, mock_request, _mock_getenv):
sample.webhook = None
expected = get_default_called_with(sample)
expected['endpoint'] = f'channels/{sample.channel_id}/messages'
assert send_message(sample.content) == sample.response
mock_request.assert_called_once_with(**expected)
def test_discord_send_message_without_content_header(sample, mock_request, _mock_getenv):
sample.content_header = None
expected = get_default_called_with(sample)
expected['json']['content'] = f'{sample.content}{sample.content_footer}'
assert send_message(sample.content) == sample.response
mock_request.assert_called_once_with(**expected)
def test_discord_send_message_without_content_footer(sample, mock_request, _mock_getenv):
sample.content_footer = None
expected = get_default_called_with(sample)
expected['json']['content'] = f'{sample.content_header}{sample.content}'
assert send_message(sample.content) == sample.response
mock_request.assert_called_once_with(**expected)
def test_discord_send_message_error(sample, mock_request, _mock_getenv):
with raises(ValueError):
sample.webhook = None
sample.channel_id = None
send_message(sample.content)
#####################################################################
def get_default_called_with(sample):
return {
'method': 'POST',
'endpoint': f'webhooks/{sample.webhook}',
'json': {
'content': f'{sample.content_header}{sample.content}{sample.content_footer}',
'username': sample.username,
'avatar_url': sample.avatar_url,
},
}
@fixture(name='mock_request', autouse=True)
def fixture_mock_request(sample):
with patch('scwrypts.http.discord.send_message.request') as mock:
mock.return_value = sample.response
yield mock
@fixture(name='_mock_getenv')
def fixture_mock_getenv(sample):
with patch('scwrypts.http.discord.send_message.getenv',) as mock:
mock.side_effect = lambda name, **kwargs: {
'DISCORD__DEFAULT_USERNAME': sample.username,
'DISCORD__DEFAULT_AVATAR_URL': sample.avatar_url,
'DISCORD__DEFAULT_WEBHOOK': sample.webhook,
'DISCORD__DEFAULT_CHANNEL_ID': sample.channel_id,
'DISCORD__CONTENT_HEADER': sample.content_header,
'DISCORD__CONTENT_FOOTER': sample.content_footer,
}[name]
yield mock

15
py/lib/scwrypts/http/linear/__init__.py
View File

@ -1 +1,14 @@
from .client import *
'''
basic scwrypts.http client for linear
configured by setting the LINEAR__API_TOKEN option in the
scwrypts environment
'''
__all__ = [
'request',
'graphql',
]
from .client import request
from .graphql import graphql

21
py/lib/scwrypts/http/linear/client.py
View File

@ -2,20 +2,11 @@ from scwrypts.env import getenv
from .. import get_request_client
REQUEST = None
def request(method, endpoint, **kwargs):
global REQUEST # pylint: disable=global-statement
if REQUEST is None:
REQUEST = get_request_client(
base_url = 'https://api.linear.app',
headers = {
'Authorization': f'bearer {getenv("LINEAR__API_TOKEN")}',
}
)
return REQUEST(method, endpoint, **kwargs)
def graphql(query):
return request('POST', 'graphql', json={'query': query})
return get_request_client(
base_url = 'https://api.linear.app',
headers = {
'Authorization': f'bearer {getenv("LINEAR__API_TOKEN")}',
},
)(method, endpoint, **kwargs)

19
py/lib/scwrypts/http/linear/conftest.py Normal file
View File

@ -0,0 +1,19 @@
from string import ascii_letters, digits
from types import SimpleNamespace
from pytest import fixture
from scwrypts.test import generate
from scwrypts.test.character_set import uri
from ..conftest import options, get_request_client_sample_data
@fixture(name='sample')
def fixture_sample():
return SimpleNamespace(
**{
**get_request_client_sample_data(),
'base_url': 'https://api.linear.app',
},
api_token = generate(str, options | {'character_set': uri}),
query = generate(str, options),
)

5
py/lib/scwrypts/http/linear/graphql.py Normal file
View File

@ -0,0 +1,5 @@
from .client import request
def graphql(query):
return request('POST', 'graphql', json={'query': query})

42
py/lib/scwrypts/http/linear/test_client.py Normal file
View File

@ -0,0 +1,42 @@
from unittest.mock import patch
from pytest import fixture
from .client import request
def test_discord_request(sample, _response):
assert _response == sample.response
def test_discord_request_client_setup(sample, mock_get_request_client, _response):
mock_get_request_client.assert_called_once_with(
base_url = sample.base_url,
headers = { 'Authorization': f'bearer {sample.api_token}' },
)
#####################################################################
@fixture(name='_response')
def fixture_response(sample):
return request(
method = sample.method,
endpoint = sample.endpoint,
**sample.payload,
)
#####################################################################
@fixture(name='mock_get_request_client', autouse=True)
def fixture_mock_get_request_client(sample):
with patch('scwrypts.http.linear.client.get_request_client') as mock:
mock.return_value = lambda method, endpoint, **kwargs: sample.response
yield mock
@fixture(name='mock_getenv', autouse=True)
def fixture_mock_getenv(sample):
with patch('scwrypts.http.linear.client.getenv',) as mock:
mock.side_effect = lambda name, **kwargs: {
'LINEAR__API_TOKEN': sample.api_token,
}[name]
yield mock

35
py/lib/scwrypts/http/linear/test_graphql.py Normal file
View File

@ -0,0 +1,35 @@
from unittest.mock import patch
from pytest import fixture
from .graphql import graphql
def test_directus_graphql(sample, _response, _mock_request):
assert _response == sample.response
def test_directus_graphql_request_payload(sample, _response, _mock_request):
_mock_request.assert_called_once_with(
'POST',
'graphql',
json = {'query': sample.query},
)
#####################################################################
@fixture(name='_response')
def fixture_response(sample, _mock_request):
return graphql(sample.query)
@fixture(name='_response_system')
def fixture_response_system(sample, _mock_request):
return graphql(sample.query)
#####################################################################
@fixture(name='_mock_request')
def fixture_mock_request(sample):
with patch('scwrypts.http.linear.graphql.request') as mock:
mock.return_value = sample.response
yield mock

55
py/lib/scwrypts/http/test_client.py Normal file
View File

@ -0,0 +1,55 @@
from unittest.mock import patch
from pytest import fixture
from .client import get_request_client
def test_request_client(sample, _response_basic):
assert _response_basic == sample.response
def test_request_client_forwards_default_headers(sample, mock_request, _response_basic):
mock_request.assert_called_once_with(
method = sample.method,
url = f'{sample.base_url}/{sample.endpoint}',
headers = sample.headers,
)
def test_get_request_client_payload(sample, _response_payload):
assert _response_payload == sample.response
def test_request_client_forwards_payload_headers(sample, mock_request, _response_payload):
assert mock_request.call_args.kwargs['headers'] == sample.headers | sample.payload_headers
#####################################################################
@fixture(name='mock_request', autouse=True)
def fixture_mock_request(sample):
with patch('scwrypts.http.client.request') as mock:
mock.return_value = sample.response
yield mock
@fixture(name='request_client', autouse=True)
def fixture_request_client(sample):
return get_request_client(sample.base_url, sample.headers)
#####################################################################
@fixture(name='_response_basic')
def fixture_response_basic(sample, request_client):
return request_client(
method = sample.method,
endpoint = sample.endpoint,
)
@fixture(name='_response_payload')
def fixture_response_payload(sample, request_client):
return request_client(
method = sample.method,
endpoint = sample.endpoint,
**{
**sample.payload,
'headers': sample.payload_headers,
},
)

66
py/lib/scwrypts/io/combined_io_stream.py
View File

@ -5,6 +5,46 @@ from sys import stdin, stdout, stderr
from scwrypts.env import getenv
@contextmanager
def get_combined_stream(input_file=None, output_file=None):
'''
context manager to open an "input_file" and "output_file"
But the "files" can be pipe-streams, stdin/stdout, or even
actual files! Helpful when trying to write CLI scwrypts
which would like to accept all kinds of input and output
configurations.
'''
with get_stream(input_file, 'r') as input_stream, get_stream(output_file, 'w+') as output_stream:
yield CombinedStream(input_stream, output_stream)
def add_io_arguments(parser, allow_input=True, allow_output=True):
'''
slap these puppies onto your argparse.ArgumentParser to
allow easy use of the get_combined_stream at the command line
'''
if allow_input:
parser.add_argument(
'-i', '--input-file',
dest = 'input_file',
default = None,
help = 'path to input file; omit for stdin',
required = False,
)
if allow_output:
parser.add_argument(
'-o', '--output-file',
dest = 'output_file',
default = None,
help = 'path to output file; omit for stdout',
required = False,
)
#####################################################################
@contextmanager
def get_stream(filename=None, mode='r', encoding='utf-8', verbose=False, **kwargs):
allowed_modes = {'r', 'w', 'w+'}
@ -34,32 +74,6 @@ def get_stream(filename=None, mode='r', encoding='utf-8', verbose=False, **kwarg
stdout.flush()
def add_io_arguments(parser, allow_input=True, allow_output=True):
if allow_input:
parser.add_argument(
'-i', '--input-file',
dest = 'input_file',
default = None,
help = 'path to input file; omit for stdin',
required = False,
)
if allow_output:
parser.add_argument(
'-o', '--output-file',
dest = 'output_file',
default = None,
help = 'path to output file; omit for stdout',
required = False,
)
@contextmanager
def get_combined_stream(input_file=None, output_file=None):
with get_stream(input_file, 'r') as input_stream, get_stream(output_file, 'w+') as output_stream:
yield CombinedStream(input_stream, output_stream)
class CombinedStream:
def __init__(self, input_stream, output_stream):
self.input = input_stream

23
py/lib/scwrypts/scwrypts/__init__.py
View File

@ -0,0 +1,23 @@
'''
scwrypts meta-configuration
provides a helpful three ways to run "scwrypts"
'scwrypts' is an agnostic, top-level executor allowing any scwrypt to be called from python workflows
'execute' is the default context set-up for python-based scwrypts
'interactive' is a context set-up for interactive, python-based scwrypts
after execution, you are dropped in a bpython shell with all the variables
configured during main() execution
'''
__all__ = [
'scwrypts',
'execute',
'interactive',
]
from .scwrypts import scwrypts
from .execute import execute
from .interactive import interactive

10
py/lib/scwrypts/scwrypts/exceptions.py
View File

@ -14,3 +14,13 @@ class MissingFlagAndEnvironmentVariableError(EnvironmentError, ArgumentError):
class MissingScwryptsExecutableError(EnvironmentError):
def __init__(self):
super().__init__(f'scwrypts must be installed and available on your PATH')
class BadScwryptsLookupError(ValueError):
def __init__(self):
super().__init__('must provide name/group/type or scwrypt lookup patterns')
class MissingScwryptsGroupOrTypeError(ValueError):
def __init__(self, group, _type):
super().__init__(f'missing required group or type (group={group} | type={_type}')

30
py/lib/scwrypts/scwrypts/scwrypts.py
View File

@ -2,28 +2,46 @@ from os import getenv
from shutil import which
from subprocess import run
from .exceptions import MissingScwryptsExecutableError
from .exceptions import MissingScwryptsExecutableError, BadScwryptsLookupError, MissingScwryptsGroupOrTypeError
def scwrypts(name, group, _type, *args, log_level=None):
def scwrypts(*args, patterns=None, name=None, group=None, _type=None, log_level=None):
'''
invoke non-python scwrypts from python
top-level scwrypts invoker from python
- patterns allows for pattern-based scwrypt lookup
- name/group/type allos for precise-match lookup
*args should be a list of strings and is forwarded to the
invoked scwrypt
see 'scwrypts --help' for more information
'''
executable = which('scwrypts')
if executable is None:
raise MissingScwryptsExecutableError()
pre_args = ''
if patterns is None and name is None:
raise BadScwryptsLookupError()
pre_args = []
if name is None:
pre_args += patterns
else:
pre_args += ['--name', name, '--group', group, '--type', _type]
if group is None or _type is None:
raise MissingScwryptsGroupOrTypeError(group, _type)
if log_level is not None:
pre_args += '--log-level {log_level}'
pre_args += ['--log-level', log_level]
depth = getenv('SUBSCWRYPT', '')
if depth != '':
depth = int(depth) + 1
return run(
f'SUBSCWRYPT={depth} {executable} --name {name} --group {group} --type {_type} {pre_args} -- {" ".join(args)}',
f'SUBSCWRYPT={depth} {executable} {pre_args} -- {" ".join(args)}',
shell=True,
executable='/bin/zsh',
check=False,

11
py/lib/scwrypts/test/__init__.py
View File

@ -1 +1,10 @@
from .random_generator import generate
'''
automated testing utilties, but primarily a random data generator
'''
__all__ = [
'generate',
]
from .generate import generate
from .character_set import *

13
py/lib/scwrypts/test/character_set.py Normal file
View File

@ -0,0 +1,13 @@
'''
string constants typically used for randomly generated data
the 'string' standard library already contains many character sets,
but not these :)
'''
__all__ = [
'uri',
]
uri = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~:/?#[]@!$&\'()*+,;='

372
py/lib/scwrypts/test/generate.py Normal file
View File

@ -0,0 +1,372 @@
from csv import writer, QUOTE_NONNUMERIC
from io import StringIO
from json import dumps, loads
from random import randint, uniform, choice
from re import sub
from string import printable
from typing import Hashable, Callable
from uuid import uuid4
from requests import Response, status_codes
from yaml import safe_dump
from .exceptions import NoDataTypeError, BadGeneratorTypeError
DEFAULT_OPTIONS = {
'data_types': None,
'minimum': 0,
'maximum': 64,
'depth': 1,
'character_set': None,
'bool_nullable': False,
'str_length': None,
'str_length_minimum': 0,
'str_length_maximum': 64,
'uuid_output_type': 'uuid', # str or 'uuid'
'list_length': 8,
'set_length': 8,
'dict_length': 8,
'csv_bool_nullable': True,
'csv_columns': None,
'csv_columns_minimum': 1,
'csv_columns_maximum': 16,
'csv_rows': None,
'csv_rows_minimum': 2,
'csv_rows_maximum': 16,
'csv_output_type': 'stringio', # str or 'stringio'
'json_initial_type': dict, # typically dict or list
'json_bool_nullable': True,
'json_output_type': 'stringio', # str or 'stringio'
'yaml_initial_type': dict, # typically dict or list
'yaml_bool_nullable': True,
'yaml_use_default_flow_style': False,
'yaml_output_type': 'stringio', # str or 'stringio'
'requests_response_status_code': status_codes.codes[200],
}
def generate(data_type=None, options=None):
'''
generate random data with the call of a function
use data_type to generate a single value
use options to set generation options (key = type, value = kwargs)
use options.data_types and omit data_type to generate a random type
'''
if options is None:
options = {}
options = DEFAULT_OPTIONS | options
if data_type is None:
if options['data_types'] is None or len(options['data_types']) == 0:
raise NoDataTypeError()
return generate(
data_type=choice(list(options['data_types'])),
options=options,
)
if not isinstance(data_type, str):
data_type = data_type.__name__
if data_type not in Generator.get_supported_data_types():
raise BadGeneratorTypeError(data_type)
return getattr(Generator, f'_{data_type}')(options)
#####################################################################
SUPPORTED_DATA_TYPES = None
class Generator:
@classmethod
def get_supported_data_types(cls):
global SUPPORTED_DATA_TYPES # pylint: disable=global-statement
if SUPPORTED_DATA_TYPES is None:
SUPPORTED_DATA_TYPES = {
sub('^_', '', data_type)
for data_type, method in Generator.__dict__.items()
if isinstance(method, staticmethod)
}
return SUPPORTED_DATA_TYPES
#####################################################################
@classmethod
def filter_data_types(cls, options, filters=None):
'''
returns an options dict with appropriately filtered data_types
if data_types are not yet defined, starts with all supported data_types
'''
if options['data_types'] is None:
options['data_types'] = Generator.get_supported_data_types()
if filters is None or len(filters) == 0:
return options
return {
**options,
'data_types': set(filter(
lambda data_type: all(( f(data_type, options) for f in filters )),
options['data_types'],
)),
}
class Filters:
@staticmethod
def hashable(data_type, _options):
if isinstance(data_type, Callable):
return isinstance(data_type(), Hashable)
if not isinstance(data_type, str):
data_type = data_type.__name__
return data_type in { 'bool', 'int', 'float', 'chr', 'str', 'uuid' }
@staticmethod
def filelike(data_type, _options):
return data_type in { 'csv', 'json', 'yaml' }
@staticmethod
def complex(data_type, _options):
return data_type in { 'requests_Response' }
@staticmethod
def basic(data_type, options):
return all([
not Generator.Filters.filelike(data_type, options),
not Generator.Filters.complex(data_type, options),
])
@staticmethod
def pythonset(data_type, _options):
if not isinstance(data_type, str):
data_type = data_type.__name__
return data_type == 'set'
@staticmethod
def csvsafe(data_type, options):
options['depth'] = max(1, options['depth'])
return all([
Generator.Filters.basic(data_type, options),
not Generator.Filters.pythonset(data_type, options),
])
@staticmethod
def jsonsafe(data_type, options):
return all([
Generator.Filters.basic(data_type, options),
not Generator.Filters.pythonset(data_type, options),
])
@staticmethod
def yamlsafe(data_type, options):
return all([
Generator.Filters.basic(data_type, options),
not Generator.Filters.pythonset(data_type, options),
])
#####################################################################
@classmethod
def get_option_with_range(cls, options, option_key, data_type=int):
'''
typically an integer range, allows both:
- setting a fixed configuration (e.g. 'str_length')
- allowing a configuration range (e.g. 'str_length_minimum' and 'str_length_maximum')
'''
fixed = options.get(option_key, None)
if fixed is not None:
return fixed
return generate(data_type, {
'minimum': options[f'{option_key}_minimum'],
'maximum': options[f'{option_key}_maximum'],
})
#####################################################################
@staticmethod
def _bool(options):
return choice([True, False, None]) if options['bool_nullable'] else choice([True, False])
@staticmethod
def _int(options):
return randint(options['minimum'], options['maximum'])
@staticmethod
def _float(options):
return uniform(options['minimum'], options['maximum'])
@staticmethod
def _chr(options):
character_set = options['character_set']
return choice(character_set) if character_set is not None else chr(randint(0,65536))
@staticmethod
def _str(options):
return ''.join((
generate(chr, options)
for _ in range(Generator.get_option_with_range(options, 'str_length'))
))
@staticmethod
def _uuid(options):
'''
creates a UUID object or a str containing a uuid (v4)
'''
uuid = uuid4()
return str(uuid) if options['uuid_output_type'] == str else uuid
@staticmethod
def _list(options):
if options['depth'] <= 0:
return []
options['depth'] -= 1
options = Generator.filter_data_types(options, [
Generator.Filters.basic,
])
return [ generate(None, {**options}) for _ in range(options['list_length']) ]
@staticmethod
def _set(options):
if options['depth'] <= 0:
return set()
options['depth'] -= 1
options = Generator.filter_data_types(options, [
Generator.Filters.hashable,
])
return { generate(None, options) for _ in range(options['set_length']) }
@staticmethod
def _dict(options):
if options['depth'] <= 0:
return {}
options['depth'] -= 1
options = Generator.filter_data_types(options, [
Generator.Filters.basic,
])
key_options = Generator.filter_data_types(options, [
Generator.Filters.hashable,
])
if len(options['data_types']) == 0 or len(key_options['data_types']) == 0:
return {}
return {
generate(None, key_options): generate(None, options)
for _ in range(options['dict_length'])
}
@staticmethod
def _csv(options):
'''
creates a StringIO object containing csv data
'''
if options['character_set'] is None:
options['character_set'] = printable
options['bool_nullable'] = options['csv_bool_nullable']
options = Generator.filter_data_types(options, [
Generator.Filters.csvsafe,
])
columns = Generator.get_option_with_range(options, 'csv_columns')
rows = Generator.get_option_with_range(options, 'csv_rows')
csv = StringIO()
csv_writer = writer(csv, quoting=QUOTE_NONNUMERIC)
options['list_length'] = columns
[ # pylint: disable=expression-not-assigned
csv_writer.writerow(generate(list, options))
for _ in range(rows)
]
csv.seek(0)
return csv.getvalue() if options['csv_output_type'] == str else csv
@staticmethod
def _json(options):
'''
creates a StringIO object or str containing json data
'''
if options['character_set'] is None:
options['character_set'] = printable
options['bool_nullable'] = options['json_bool_nullable']
options['uuid_output_type'] = str
options = Generator.filter_data_types(options, [
Generator.Filters.jsonsafe,
])
json = dumps(generate(
options['json_initial_type'],
{**options},
))
return json if options['json_output_type'] == str else StringIO(json)
@staticmethod
def _yaml(options):
'''
creates a StringIO object or str containing yaml data
'''
if options['character_set'] is None:
options['character_set'] = printable
options['bool_nullable'] = options['yaml_bool_nullable']
options['uuid_output_type'] = str
options = Generator.filter_data_types(options, [
Generator.Filters.yamlsafe,
])
yaml = StringIO()
safe_dump(
generate(options['yaml_initial_type'], {**options}),
yaml,
default_flow_style=options['yaml_use_default_flow_style'],
)
yaml.seek(0)
return yaml.getvalue() if options['yaml_output_type'] == str else yaml
@staticmethod
def _requests_Response(options):
'''
creates a requests.Response-like object containing json data
'''
options['json_output_type'] = str
response = Response()
response.status_code = options['requests_response_status_code']
json = loads(generate('json', options))
response.json = lambda: json
return response

240
py/lib/scwrypts/test/random_generator.py
View File

@ -1,240 +0,0 @@
from csv import writer, QUOTE_NONNUMERIC
from io import StringIO
from json import dumps
from random import randint, uniform, choice
from re import sub
from string import printable
from yaml import safe_dump
from .exceptions import NoDataTypeError, BadGeneratorTypeError
SUPPORTED_DATA_TYPES = None
DEFAULT_OPTIONS = {
'data_types': None,
'minimum': 0,
'maximum': 64,
'depth': 1,
'character_set': None,
'bool_nullable': False,
'str_length': None,
'str_minimum_length': 0,
'str_maximum_length': 32,
'list_length': 8,
'set_length': 8,
'dict_length': 8,
'dict_key_types': {int, float, chr, str},
'csv_columns': None,
'csv_columns_minimum': 1,
'csv_columns_maximum': 16,
'csv_rows': None,
'csv_rows_minimum': 2,
'csv_rows_maximum': 16,
'json_initial_type': dict,
'yaml_initial_type': dict,
}
def generate(data_type=None, options=None):
'''
generate random data with the call of a function
use data_type to generate a single value
use options to set generation options (key = type, value = kwargs)
use options.data_types and omit data_type to generate a random type
'''
if options is None:
options = {**DEFAULT_OPTIONS}
else:
options = DEFAULT_OPTIONS | options
if data_type is None and options['data_types'] is None:
raise NoDataTypeError()
if data_type is None and options['data_types'] is not None:
return generate(data_type=choice(list(options['data_types'])), options=options)
if not isinstance(data_type, str):
data_type = data_type.__name__
if data_type not in Generator.get_supported_data_types():
raise BadGeneratorTypeError(data_type)
return getattr(Generator, f'_{data_type}')(options)
#####################################################################
class Generator:
@classmethod
def get_supported_data_types(cls):
global SUPPORTED_DATA_TYPES # pylint: disable=global-statement
if SUPPORTED_DATA_TYPES is None:
SUPPORTED_DATA_TYPES = {
sub('^_', '', data_type)
for data_type, method in Generator.__dict__.items()
if isinstance(method, staticmethod)
}
return SUPPORTED_DATA_TYPES
@staticmethod
def _bool(options):
return choice([True, False, None]) if options['bool_nullable'] else choice([True, False])
@staticmethod
def _int(options):
return randint(options['minimum'], options['maximum'])
@staticmethod
def _float(options):
return uniform(options['minimum'], options['maximum'])
@staticmethod
def _chr(options):
character_set = options['character_set']
return choice(character_set) if character_set is not None else chr(randint(0,65536))
@staticmethod
def _str(options):
length = options['str_length']
if length is None:
length = generate(int, {
'minimum': options['str_minimum_length'],
'maximum': options['str_maximum_length'],
})
return ''.join((generate(chr, options) for _ in range(length)))
@staticmethod
def _list(options):
if options['depth'] <= 0:
return []
options['depth'] -= 1
if options['data_types'] is None:
options['data_types'] = {bool, int, float, chr, str}
return [ generate(None, options) for _ in range(options['list_length']) ]
@staticmethod
def _set(options):
if options['depth'] <= 0:
return set()
options['depth'] -= 1
if options['data_types'] is None:
options['data_types'] = {bool, int, float, chr, str}
set_options = options | {'data_types': options['data_types'] - {list, dict, set}}
return { generate(None, set_options) for _ in range(options['set_length']) }
@staticmethod
def _dict(options):
if options['depth'] <= 0:
return {}
options['depth'] -= 1
if options['data_types'] is None:
options['data_types'] = {bool, int, float, chr, str, list, set, dict}
if len(options['data_types']) == 0:
return {}
key_options = options | {'data_types': options['dict_key_types']}
return {
generate(None, key_options): generate(None, options)
for _ in range(options['dict_length'])
}
@staticmethod
def _csv(options):
'''
creates a StringIO object containing csv data
'''
if options['data_types'] is None:
options['data_types'] = {int, float, str}
columns = options['csv_columns']
if columns is None:
columns = max(1, generate(int, {
'minimum': options['csv_columns_minimum'],
'maximum': options['csv_columns_maximum'],
}))
rows = options['csv_rows']
if rows is None:
rows = max(1, generate(int, {
'minimum': options['csv_rows_minimum'],
'maximum': options['csv_rows_maximum'],
}))
if options['character_set'] is None:
options['character_set'] = printable
csv = StringIO()
csv_writer = writer(csv, quoting=QUOTE_NONNUMERIC)
options['list_length'] = columns
for line in [ generate(list, options) for _ in range(rows) ]:
csv_writer.writerow(line)
csv.seek(0)
return csv
@staticmethod
def _json(options):
'''
creates a str containing json data
'''
if options['data_types'] is None:
options['data_types'] = {bool, int, float, str, list, dict}
if options['character_set'] is None:
options['character_set'] = printable
options['dict_key_types'] = { int, float, str }
data = generate(options['json_initial_type'], options)
return dumps(data)
@staticmethod
def _yaml(options):
'''
creates a StringIO object containing yaml data
'''
if options['data_types'] is None:
options['data_types'] = {bool, int, float, str, list, dict}
if options['character_set'] is None:
options['character_set'] = printable
options['dict_key_types'] = { int, float, str }
yaml = StringIO()
safe_dump(generate(options['yaml_initial_type'], options), yaml, default_flow_style=False)
yaml.seek(0)
return yaml
#####################################################################
if __name__ == '__main__':
print(generate('json', {'depth': 3}))

18
py/lib/scwrypts/test/test_random_generator.py → py/lib/scwrypts/test/test_generate.py
View File

@ -1,14 +1,24 @@
from os import getenv
from pprint import pprint
from random import randint
from .random_generator import generate, Generator
from .generate import generate, Generator
ITERATIONS = int(
getenv(
'PYTEST_ITERATIONS__scwrypts__test__generator',
getenv('PYTEST_ITERATIONS', '99'), # CI should use at least 999
)
)
ITERATIONS = int(getenv('PYTEST_ITERATIONS__scwrypts__test__random_generator', getenv('PYTEST_ITERATIONS', '999')))
FILE_LIKE_DATA_TYPES = { 'csv', 'json', 'yaml' }
def test_generate(): # generators should be quick and "just work" (no Exceptions)
print()
for data_type in Generator.get_supported_data_types():
print(f'------- {data_type} -------')
sample = generate(data_type)
pprint(sample.getvalue() if data_type in {'csv', 'json', 'yaml'} else sample)
for _ in range(ITERATIONS):
generate(data_type)
@ -41,4 +51,4 @@ def test_generate_range_negative():
def test_generate_bool_nullable():
for data_type in Generator.get_supported_data_types():
generate(data_type, {'bool': {'nullable': True}})
generate(data_type, {'bool_nullable': True})

11
py/lib/scwrypts/twilio/__init__.py
View File

@ -1,2 +1,13 @@
'''
loads the twilio.rest.Client by referencing TWILIO__API_KEY,
TWILIO__API_SECRET, and TWILIO__ACCOUNT_SID in your scwrypts
environment
'''
__all__ = [
'get_client',
'send_sms',
]
from .client import get_client
from .send_sms import send_sms