<commit_before># This file is part of khmer, https://github.com/dib-lab/khmer/, and is
# Copyright (C) 2013-2015, Michigan State University.
# Copyright (C) 2015-2016, The Regents of the University of California.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
#       notice, this list of conditions and the following disclaimer.
#
#     * Redistributions in binary form must reproduce the above
#       copyright notice, this list of conditions and the following
#       disclaimer in the documentation and/or other materials provided
#       with the distribution.
#
#     * Neither the name of the Michigan State University nor the names
#       of its contributors may be used to endorse or promote products
#       derived from this software without specific prior written
#       permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Contact: <EMAIL>
# pylint: disable=missing-docstring,invalid-name

# Tests for the ReadParser and Read classes.
from __future__ import print_function
from __future__ import absolute_import
from khmer import _Countgraph, _Counttable, _SmallCountgraph, _SmallCounttable
from khmer import _Nodegraph, _Nodetable
from khmer import ReadParser, Counttable, Nodegraph
from screed import Record
from . import khmer_tst_utils as utils
import pytest
from functools import reduce  # pylint: disable=redefined-builtin


PRIMES_1m = [1000003, 1009837]


# all the table types!
@pytest.fixture(params=[_Countgraph, _Counttable, _SmallCountgraph,
                        _SmallCounttable, _Nodegraph, _Nodetable])
def tabletype(request):
    return request.param

def test_read_cleaning_consume_seqfile(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    x.consume_seqfile(infile)

    # the relevant read will automatically get uppercased => abundance of 2
    kmer = "caggcgcccaccacc".upper()
    assert x.get(kmer) == 2

    # the 2nd read with this k-mer in it has an N in it.
    kmer = "CCTCATCGGCACCAG"
    assert x.get(kmer) == 1               # this should be 2 in the future

    # the 2nd read with this k-mer in it has a Z in it
    kmer = "ACTGAGCTTCATGTC"
    assert x.get(kmer) == 1               # this should be 2 in the future


def test_read_cleaning_consume_read_by_read(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    for read in ReadParser(infile):
        x.consume(read.sequence)          # consume raw sequence

    # the relevant read will be entirely ignored
    # (b/c ReadParser does not uppercase)
    kmer = "caggcgcccaccacc".upper()
    assert x.get(kmer) == 1

    # consume will ignore the invalid base in 2nd read containing this k-mer,
    # so the k-mer will have an abundance of 2.
    kmer = "CCTCATCGGCACCAG"
    assert x.get(kmer) == 2

    # consume will ignore the invalid base in 2nd read containing this k-mer,
    # so the k-mer will have an abundance of 2.
    kmer = "ACTGAGCTTCATGTC"
    assert x.get(kmer) == 2


def test_read_cleaning_consume_read_by_read_cleaned_seq(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    for read in ReadParser(infile):
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    # the relevant read will be cleaned & loaded
    kmer = "caggcgcccaccacc".upper()
    assert x.get(kmer) == 2

    # this k-mer will be correctly loaded
    kmer = "CCTCATCGGCACCAG"
    assert x.get(kmer) == 2

    # this k-mer will be correctly loaded
    kmer = "ACTGAGCTTCATGTC"
    assert x.get(kmer) == 2


def test_read_cleaning_abundance_distribution(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    y = _Nodegraph(15, PRIMES_1m)

    x.consume_seqfile(infile)

    dist = x.abundance_distribution(infile, y)
    assert dist[1] == 41
    assert dist[2] == 42


def test_read_cleaning_trim_functions_lowercase(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    # read this in using "approved good" behavior w/cleaned_seq
    x = tabletype(8, PRIMES_1m)
    for read in ReadParser(infile):
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    s = "caggcgcccaccaccgtgccctccaacctgatggt"
    _, where = x.trim_on_abundance(s, 1)
    assert where == 35                    # in future, should be '0'

    _, where = x.trim_below_abundance(s, 2)
    assert where == 35                    # in future, should be ?? @CTB

    posns = x.find_spectral_error_positions(s, 1)
    assert posns == []                    # in future, should do same


def test_read_cleaning_trim_functions_N(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    # read this in using "approved good" behavior w/cleaned_seq
    x = tabletype(8, PRIMES_1m)
    for read in ReadParser(infile):
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    s = "ACTGGGCGTAGNCGGTGTCCTCATCGGCACCAGC"
    _, where = x.trim_on_abundance(s, 1)
    assert where == 0                     # in future, should be ??

    _, where = x.trim_below_abundance(s, 2)
    assert where == 0                     # in future, should be ??

    with pytest.raises(ValueError):
        posns = x.find_spectral_error_positions(s, 1)
    # assert posns == []                    # in future, should return []


def test_read_cleaning_trim_functions_bad_dna(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    # read this in using "approved good" behavior w/cleaned_seq
    x = tabletype(8, PRIMES_1m)
    for read in ReadParser(infile):
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    s = "CCGGCGTGGTTZZYAGGTCACTGAGCTTCATGTC"
    _, where = x.trim_on_abundance(s, 1)
    assert where == 0                     # in future, should be ??

    _, where = x.trim_below_abundance(s, 2)
    assert where == 0                     # in future, should be ??

    with pytest.raises(ValueError):
        posns = x.find_spectral_error_positions(s, 1)
    # assert posns == []                    # in future, should return [11]<commit_msg>Add a fixture for ReadParser use<commit_after># This file is part of khmer, https://github.com/dib-lab/khmer/, and is
# Copyright (C) 2013-2015, Michigan State University.
# Copyright (C) 2015-2016, The Regents of the University of California.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
#       notice, this list of conditions and the following disclaimer.
#
#     * Redistributions in binary form must reproduce the above
#       copyright notice, this list of conditions and the following
#       disclaimer in the documentation and/or other materials provided
#       with the distribution.
#
#     * Neither the name of the Michigan State University nor the names
#       of its contributors may be used to endorse or promote products
#       derived from this software without specific prior written
#       permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Contact: <EMAIL>
# pylint: disable=missing-docstring,invalid-name

# Tests for the ReadParser and Read classes.
from __future__ import print_function
from __future__ import absolute_import
from khmer import _Countgraph, _Counttable, _SmallCountgraph, _SmallCounttable
from khmer import _Nodegraph, _Nodetable
from khmer import ReadParser, Nodegraph
from . import khmer_tst_utils as utils
import pytest


PRIMES_1m = [1000003, 1009837]


# all the table types!
@pytest.fixture(params=[_Countgraph, _Counttable, _SmallCountgraph,
                        _SmallCounttable, _Nodegraph, _Nodetable])
def tabletype(request):
    return request.param


@pytest.fixture
def reads():
    infile = utils.get_test_data('valid-read-testing.fq')
    reads = ReadParser(infile)
    yield reads
    reads.close()


def test_read_cleaning_consume_seqfile(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')
    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    x.consume_seqfile(infile)

    # the relevant read will automatically get uppercased => abundance of 2
    kmer = "caggcgcccaccacc".upper()
    assert x.get(kmer) == 2

    # the 2nd read with this k-mer in it has an N in it.
    kmer = "CCTCATCGGCACCAG"
    assert x.get(kmer) == 1               # this should be 2 in the future

    # the 2nd read with this k-mer in it has a Z in it
    kmer = "ACTGAGCTTCATGTC"
    assert x.get(kmer) == 1               # this should be 2 in the future


def test_read_cleaning_consume_read_by_read(tabletype, reads):
    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    for read in reads:
        x.consume(read.sequence)          # consume raw sequence

    # the relevant read will be entirely ignored
    # (b/c ReadParser does not uppercase)
    kmer = "caggcgcccaccacc".upper()
    assert x.get(kmer) == 1

    # consume will ignore the invalid base in 2nd read containing this k-mer,
    # so the k-mer will have an abundance of 2.
    kmer = "CCTCATCGGCACCAG"
    assert x.get(kmer) == 2

    # consume will ignore the invalid base in 2nd read containing this k-mer,
    # so the k-mer will have an abundance of 2.
    kmer = "ACTGAGCTTCATGTC"
    assert x.get(kmer) == 2


def test_read_cleaning_consume_read_by_read_cleaned_seq(tabletype, reads):
    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    for read in reads:
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    # the relevant read will be cleaned & loaded
    kmer = "caggcgcccaccacc".upper()
    assert x.get(kmer) == 2

    # this k-mer will be correctly loaded
    kmer = "CCTCATCGGCACCAG"
    assert x.get(kmer) == 2

    # this k-mer will be correctly loaded
    kmer = "ACTGAGCTTCATGTC"
    assert x.get(kmer) == 2


def test_read_cleaning_abundance_distribution(tabletype):
    infile = utils.get_test_data('valid-read-testing.fq')

    if tabletype == _Nodegraph or tabletype == _Nodetable:
        return

    x = tabletype(15, PRIMES_1m)
    y = _Nodegraph(15, PRIMES_1m)

    x.consume_seqfile(infile)

    dist = x.abundance_distribution(infile, y)
    assert dist[1] == 41
    assert dist[2] == 42


def test_read_cleaning_trim_functions_lowercase(tabletype, reads):
    # read this in using "approved good" behavior w/cleaned_seq
    x = tabletype(8, PRIMES_1m)
    for read in reads:
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    s = "caggcgcccaccaccgtgccctccaacctgatggt"
    _, where = x.trim_on_abundance(s, 1)
    assert where == 35                    # in future, should be '0'

    _, where = x.trim_below_abundance(s, 2)
    assert where == 35                    # in future, should be ?? @CTB

    posns = x.find_spectral_error_positions(s, 1)
    assert posns == []                    # in future, should do same


def test_read_cleaning_trim_functions_N(tabletype, reads):
    # read this in using "approved good" behavior w/cleaned_seq
    x = tabletype(8, PRIMES_1m)
    for read in reads:
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    s = "ACTGGGCGTAGNCGGTGTCCTCATCGGCACCAGC"
    _, where = x.trim_on_abundance(s, 1)
    assert where == 0                     # in future, should be ??

    _, where = x.trim_below_abundance(s, 2)
    assert where == 0                     # in future, should be ??

    with pytest.raises(ValueError):
        posns = x.find_spectral_error_positions(s, 1)
    # assert posns == []                    # in future, should return []


def test_read_cleaning_trim_functions_bad_dna(tabletype, reads):
    # read this in using "approved good" behavior w/cleaned_seq
    x = tabletype(8, PRIMES_1m)
    for read in reads:
        x.consume(read.cleaned_seq)       # consume cleaned_seq

    s = "CCGGCGTGGTTZZYAGGTCACTGAGCTTCATGTC"
    _, where = x.trim_on_abundance(s, 1)
    assert where == 0                     # in future, should be ??

    _, where = x.trim_below_abundance(s, 2)
    assert where == 0                     # in future, should be ??

    with pytest.raises(ValueError):
        posns = x.find_spectral_error_positions(s, 1)
    # assert posns == []                    # in future, should return [11]
{'fraction_non_alphanumeric': 0.07936397722538537, 'fraction_numerical': 0.01333148173864741, 'mean_word_length': 3.0813261547180506, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1765365', 'n_tokens_mistral': 4820, 'n_tokens_neox': 4416, 'n_words': 1460}