Skip to content
Snippets Groups Projects
Commit 0a5fc5e1 authored by Alex Diebold's avatar Alex Diebold
Browse files

moved argparse code into it's own method in db.py for easier use in testing...

moved argparse code into it's own method in db.py for easier use in testing code. working on unit tests. making progress on _query() testing
parent 8a3447f5
No related branches found
No related tags found
No related merge requests found
......@@ -526,7 +526,7 @@ def _create(args):
session = Session()
#make n/a experiment
na_exp = Experiment(name='n/a', description='default experiment -- not a part of any experiment')
na_exp = Experiment(name='n/a', desc='default experiment -- not a part of any experiment')
na_exp.experiment_id = 1
session.add(na_exp)
session.commit()
......@@ -777,14 +777,16 @@ def _query(args):
#initial Instrument query
init_inst_query = session.query(Instrument)
#check if there are any inst names to use to filter
if query_in[C.INST][C.NAME][0]:
#filter using name(s)
if len(query_in[C.INST][C.NAME][0].replace(' ', '')) > 0:
print(len(query_in[C.INST][C.NAME][0].replace(' ', '')))
print(query_in[C.INST][C.NAME][0].replace(' ', ''))
#filter using name(s
for i_name in query_in[C.INST][C.NAME]:
queries[C.INST].append(init_inst_query.filter(Instrument.name == i_name))
else:
queries[C.INST].append(init_inst_query)
#check if specific site(s) was/were specified
if query_in[C.INST][C.SITE][0]:
if len(query_in[C.INST][C.SITE][0].replace(' ', '')) > 0:
#make copy of current queries
temp = queries[C.INST].copy()
#clear out current queries
......@@ -794,7 +796,7 @@ def _query(args):
for i_site in query_in[C.INST][C.SITE]:
queries[C.INST].append(qry.filter(Instrument.site.has(name = i_site)))
# check if specific filetype(s) was/were specified
if query_in[C.INST][C.FILETYPE][0]:
if len(query_in[C.INST][C.FILETYPE][0].replace(' ', '')) > 0:
#make copy of current queries
temp = queries[C.INST].copy()
#clear out current queries
......@@ -1044,6 +1046,8 @@ def _query(args):
print('{} - {}'.format(num, val))
num += 1
return queries
'''add column to database
optional arguments:
......@@ -1057,7 +1061,6 @@ optional arguments:
format as '-t pattern.interval'
-f FILE, --file FILE format as '-f instrument.experiment.filetype.relative
_path.start_time.end_time'
**currently not in use**
'''
......@@ -1385,7 +1388,16 @@ def _test(args):
else:
raise FileNotFoundError('database \'{}\' does not exist'.format(args.filename))
def main():
'''parse arguments
takes arguments from sys.argv[1:] or a list of strings
ex. ['test.db', 'query', '-i', '-s', '-e']
returns the resulting parser
'''
def parse_args(args):
parser = argparse.ArgumentParser()
......@@ -1449,12 +1461,22 @@ def main():
parser_test.add_argument('-c', '--testing_create', help='testing _create() functionality with yaml')
'''
args = parser.parse_args()
return parser.parse_args(args)
def main():
print(sys.argv[1:])
#skip the program name, pass on rest of arguments
args = parse_args(sys.argv[1:])
#args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG)
#run the corresponding method that was called
try:
return args.func(args)
#database doesn't exist or value given is incorrect
except (FileNotFoundError, ValueError) as e:
LOG.error(str(e))
raise
......
from db import Instrument, Site, Experiment, FileType, File
from db import parse_args
from db import C
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import module
import logging
import os
def setup_function(function):
pass
LOG = logging.getLogger(__name__)
DATABASE = 'test.db'
'''set up logging to debug mode at the very start
'''
def setup_module(m):
logging.basicConfig(level=logging.DEBUG)
class TestCreate:
def setup(self):
pass
def teardown(self):
pass
def setup_class(cls):
pass
def teardown_class(cls):
pass
def setup_method(self, method):
pass
def teardown_method(self, method):
pass
class TestSync:
def setup(self):
pass
def teardown(self):
pass
def setup_class(cls):
pass
def teardown_class(cls):
pass
def setup_method(self, method):
pass
def teardown_method(self, method):
pass
class TestQuery:
def setup(self):
pass
def teardown(self):
pass
'''create a fresh database to use for testing
'''
def setup_class(cls):
#remove database if it is already made to start fresh
if os.path.isfile(DATABASE):
os.remove(DATABASE)
#simulate 'python db.py test.db create'
args = parse_args([DATABASE, 'create'])
try:
return args.func(args)
except (FileNotFoundError, ValueError) as e:
LOG.error(str(e))
raise
'''remove database after query testing is complete
'''
def teardown_class(cls):
os.remove(DATABASE)
def setup_method(self, method):
pass
def teardown_method(self, method):
#module.input = input
pass
def test_inst_1(self):
engine = create_engine('sqlite:///' + DATABASE)
Session = sessionmaker(bind=engine)
session = Session()
#monkeypatch.setitem('builtins.input', lambda x: '\n\n\n')
module.input = lambda: '\n\n\n'
#parse_args.input = ''
#simulate 'python db.py test.db query -i'
args = parse_args([DATABASE, 'query', '-i'])
query_result = args.func(args)[C.INST]
#make the query that should be output
query_expected = session.query(Instrument)
assert(query_result == query_expected)
class TestAdd:
def setup(self):
pass
def teardown(self):
pass
def setup_class(cls):
pass
def teardown_class(cls):
pass
def setup_method(self, method):
pass
def teardown_method(self, method):
pass
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment