# Path: taxtastic/ncbi.py
# def read_names(rows, source_id=1):
#     """Return an iterator of rows ready to insert into table
#     "names". Adds columns "is_primary" (identifying the primary name
#     for each tax_id with a vaule of 1) and "is_classified" (always None).
# 
#     * rows - iterator of lists (eg, output from read_archive or read_dmp)
#     * unclassified_regex - a compiled re matching "unclassified" names
# 
#     From the NCBI docs:
# 
#     Taxonomy names file (names.dmp):
#         tax_id -- the id of node associated with this name
#         name_txt -- name itself
#         unique name -- the unique variant of this name if name not unique
#         name class -- (synonym, common name, ...)
# 
#     """
# 
#     ncbi_keys = ['tax_id', 'tax_name', 'unique_name', 'name_class']
#     extra_keys = ['source_id', 'is_primary', 'is_classified']
# 
#     # is_classified applies to species only; we will set this value
#     # later
#     is_classified = None
# 
#     tax_id = ncbi_keys.index('tax_id')
#     tax_name = ncbi_keys.index('tax_name')
#     unique_name = ncbi_keys.index('unique_name')
#     name_class = ncbi_keys.index('name_class')
# 
#     yield ncbi_keys + extra_keys
# 
#     for tid, grp in itertools.groupby(rows, itemgetter(tax_id)):
#         # confirm that each tax_id has exactly one scientific name
#         num_primary = 0
#         for r in grp:
#             is_primary = r[name_class] == 'scientific name'
#             # fix primary key uniqueness violation
#             if r[unique_name]:
#                 r[tax_name] = r[unique_name]
#             num_primary += is_primary
#             yield (r + [source_id, is_primary, is_classified])
# 
#         assert num_primary == 1
# 
# def read_archive(archive, fname):
#     """Return an iterator of unique rows from a zip archive.
# 
#     * archive - path to the zip archive.
#     * fname - name of the compressed file within the archive.
# 
#     """
# 
#     # Note that deduplication here is equivalent to an upsert/ignore,
#     # but avoids requirement for a database-specific implementation.
# 
#     zfile = zipfile.ZipFile(archive)
#     contents = zfile.open(fname, 'r')
#     fobj = io.TextIOWrapper(contents)
# 
#     seen = set()
#     for line in fobj:
#         line = line.rstrip('\t|\n')
#         if line not in seen:
#             yield line.split('\t|\t')
#             seen.add(line)
# 
# Path: tests/config.py
# class TestBase(unittest.TestCase):
#     """
#     Base class for unit tests
#     """
# 
#     outputdir = outputdir
# 
#     def outdir(self):
#         """
#         Name an outputdir as outpudir/module.class.method
#         """
#         funcname = '.'.join(self.id().split('.')[-3:])
#         return path.join(self.outputdir, funcname)
# 
#     def mkoutdir(self, clobber=True):
#         """
#         Create output directory (destructively if clobber is True)
#         """
# 
#         outdir = self.outdir()
#         mkdir(outdir, clobber)
#         return outdir
# 
#     def suppress_stdout(self):
#         self.old_stdout = sys.stdout
#         sys.stdout = StringIO()
# 
#     def suppress_stderr(self):
#         self.old_stderr = sys.stderr
#         sys.stderr = StringIO()
# 
#     def tearDown(self):
#         if hasattr(self, 'old_stdout'):
#             sys.stdout = self.old_stdout
#         if hasattr(self, 'old_stderr'):
#             sys.stdout = self.old_stderr


import re
import os
import logging
import sqlalchemy
import taxtastic
import taxtastic.ncbi
from os import path
from taxtastic.ncbi import read_names, read_archive
from . import config
from .config import TestBase
#!/usr/bin/env python




log = logging

outputdir = config.outputdir
datadir = config.datadir
ncbi_master_db = config.ncbi_master_db
ncbi_data = config.ncbi_data


class TestDbconnect(TestBase):

    def test01(self):
        engine = sqlalchemy.create_engine('sqlite:///' + ncbi_master_db)
        taxtastic.ncbi.db_connect(engine)
        with engine.begin() as con:
            result = con.execute(
                'select name from sqlite_master where type = "table"')
            tables = set(i[0] for i in result)
            self.assertTrue(
                set(['nodes', 'names', 'merged', 'source']).issubset(tables))



# Path: taxtastic/ncbi.py
# def read_names(rows, source_id=1):
#     """Return an iterator of rows ready to insert into table
#     "names". Adds columns "is_primary" (identifying the primary name
#     for each tax_id with a vaule of 1) and "is_classified" (always None).
# 
#     * rows - iterator of lists (eg, output from read_archive or read_dmp)
#     * unclassified_regex - a compiled re matching "unclassified" names
# 
#     From the NCBI docs:
# 
#     Taxonomy names file (names.dmp):
#         tax_id -- the id of node associated with this name
#         name_txt -- name itself
#         unique name -- the unique variant of this name if name not unique
#         name class -- (synonym, common name, ...)
# 
#     """
# 
#     ncbi_keys = ['tax_id', 'tax_name', 'unique_name', 'name_class']
#     extra_keys = ['source_id', 'is_primary', 'is_classified']
# 
#     # is_classified applies to species only; we will set this value
#     # later
#     is_classified = None
# 
#     tax_id = ncbi_keys.index('tax_id')
#     tax_name = ncbi_keys.index('tax_name')
#     unique_name = ncbi_keys.index('unique_name')
#     name_class = ncbi_keys.index('name_class')
# 
#     yield ncbi_keys + extra_keys
# 
#     for tid, grp in itertools.groupby(rows, itemgetter(tax_id)):
#         # confirm that each tax_id has exactly one scientific name
#         num_primary = 0
#         for r in grp:
#             is_primary = r[name_class] == 'scientific name'
#             # fix primary key uniqueness violation
#             if r[unique_name]:
#                 r[tax_name] = r[unique_name]
#             num_primary += is_primary
#             yield (r + [source_id, is_primary, is_classified])
# 
#         assert num_primary == 1
# 
# def read_archive(archive, fname):
#     """Return an iterator of unique rows from a zip archive.
# 
#     * archive - path to the zip archive.
#     * fname - name of the compressed file within the archive.
# 
#     """
# 
#     # Note that deduplication here is equivalent to an upsert/ignore,
#     # but avoids requirement for a database-specific implementation.
# 
#     zfile = zipfile.ZipFile(archive)
#     contents = zfile.open(fname, 'r')
#     fobj = io.TextIOWrapper(contents)
# 
#     seen = set()
#     for line in fobj:
#         line = line.rstrip('\t|\n')
#         if line not in seen:
#             yield line.split('\t|\t')
#             seen.add(line)
# 
# Path: tests/config.py
# class TestBase(unittest.TestCase):
#     """
#     Base class for unit tests
#     """
# 
#     outputdir = outputdir
# 
#     def outdir(self):
#         """
#         Name an outputdir as outpudir/module.class.method
#         """
#         funcname = '.'.join(self.id().split('.')[-3:])
#         return path.join(self.outputdir, funcname)
# 
#     def mkoutdir(self, clobber=True):
#         """
#         Create output directory (destructively if clobber is True)
#         """
# 
#         outdir = self.outdir()
#         mkdir(outdir, clobber)
#         return outdir
# 
#     def suppress_stdout(self):
#         self.old_stdout = sys.stdout
#         sys.stdout = StringIO()
# 
#     def suppress_stderr(self):
#         self.old_stderr = sys.stderr
#         sys.stderr = StringIO()
# 
#     def tearDown(self):
#         if hasattr(self, 'old_stdout'):
#             sys.stdout = self.old_stdout
#         if hasattr(self, 'old_stderr'):
#             sys.stdout = self.old_stderr


# Path: tests/test_ncbi.py
import re
import os
import logging
import sqlalchemy
import taxtastic
import taxtastic.ncbi
from os import path
from taxtastic.ncbi import read_names, read_archive
from . import config
from .config import TestBase

#!/usr/bin/env python




log = logging

outputdir = config.outputdir
datadir = config.datadir
ncbi_master_db = config.ncbi_master_db
ncbi_data = config.ncbi_data


class TestDbconnect(TestBase):

    def test01(self):
        engine = sqlalchemy.create_engine('sqlite:///' + ncbi_master_db)
        taxtastic.ncbi.db_connect(engine)
        with engine.begin() as con:
            result = con.execute(
                'select name from sqlite_master where type = "table"')
            tables = set(i[0] for i in result)
            self.assertTrue(
                set(['nodes', 'names', 'merged', 'source']).issubset(tables))



class TestLoadData(TestBase):