<commit_before>#!/usr/bin/env python3

"""
test_correct_splice_graph.py: tests for the exfi.correct_splice_graph submodule
"""


from unittest import TestCase, main

from tempfile import \
    mkstemp

from os import remove

import networkx as nx

from Bio import \
    SeqIO

from exfi.build_baited_bloom_filter import \
    build_baited_bloom_filter

from exfi.find_exons import \
    _find_exons_pipeline

from exfi.build_splice_graph import \
    build_splice_graph

from exfi.correct_splice_graph import \
    _prepare_sealer, \
    _run_sealer, \
    _collect_sealer_results, \
    _sculpt_graph, \
    correct_splice_graph



from tests.auxiliary_functions import \
    CustomAssertions


def _comopose_args(bloom_fn, gfa_fn):
    """Compose a dict of args with two variables"""
    return {
        "kmer": 30,
        "input_bloom": bloom_fn,
        "bloom_size": "500M",
        "levels": 1,
        "input_fasta": "tests/correct_splice_graph/transcript.fa",
        "max_fp_bases": 5,
        "max_overlap": 10,
        "output_gfa": gfa_fn,
        "threads": 4,
        "max_gap_size": 10,
        "reads": ["tests/correct_splice_graph/reads.fa"],
        "output_bloom" : bloom_fn,
        "bloom_filter": bloom_fn
    }


TEMP_BLOOM = mkstemp()
TEMP_GFA = mkstemp()
ARGS = _comopose_args(TEMP_BLOOM[1], TEMP_GFA[1])
build_baited_bloom_filter(ARGS)
POSITIVE_EXONS_BED = list(_find_exons_pipeline(ARGS))
SPLICE_GRAPH = build_splice_graph(POSITIVE_EXONS_BED)


def tearDownModule():
    """Remove temporary bloom and temporary GFA files"""
    # pylint: disable=invalid-name
    remove(TEMP_BLOOM[1])
    remove(TEMP_GFA[1])



class TestPrepareSealer(TestCase, CustomAssertions):
    """_prepare_sealer(splice_graph, args)
    (nx.DiGraph, dict_of_parameters) -> str
    """

    def test_file_creation(self):
        """exfi.correct_splice_graph._prepare_sealer: test creation"""
        sealer_input_fn = _prepare_sealer(SPLICE_GRAPH, ARGS)
        actual = list(SeqIO.parse(sealer_input_fn, format="fasta"))
        expected = list(SeqIO.parse("tests/correct_splice_graph/to_seal.fa", format="fasta"))
        remove(sealer_input_fn)
        self.assertEqualListOfSeqrecords(actual, expected)


class TestRunSealer(TestCase, CustomAssertions):
    """_run_sealer(sealer_input_fn, args):
    (str, dict) -> str
    """

    def test_run(self):<commit_msg>All tmp data is cleaned!<commit_after>#!/usr/bin/env python3

"""
test_correct_splice_graph.py: tests for the exfi.correct_splice_graph submodule
"""


from unittest import TestCase, main

from tempfile import \
    mkstemp

from os import remove

import networkx as nx

from Bio import \
    SeqIO

from exfi.build_baited_bloom_filter import \
    build_baited_bloom_filter

from exfi.find_exons import \
    _find_exons_pipeline

from exfi.build_splice_graph import \
    build_splice_graph

from exfi.correct_splice_graph import \
    _prepare_sealer, \
    _run_sealer, \
    _collect_sealer_results, \
    _sculpt_graph, \
    correct_splice_graph



from tests.auxiliary_functions import \
    CustomAssertions


def _comopose_args(bloom_fn, gfa_fn):
    """Compose a dict of args with two variables"""
    return {
        "kmer": 30,
        "input_bloom": bloom_fn,
        "bloom_size": "500M",
        "levels": 1,
        "input_fasta": "tests/correct_splice_graph/transcript.fa",
        "max_fp_bases": 5,
        "max_overlap": 10,
        "output_gfa": gfa_fn,
        "threads": 4,
        "max_gap_size": 10,
        "reads": ["tests/correct_splice_graph/reads.fa"],
        "output_bloom" : bloom_fn,
        "bloom_filter": bloom_fn
    }


TEMP_BLOOM = mkstemp()
TEMP_GFA = mkstemp()
ARGS = _comopose_args(TEMP_BLOOM[1], TEMP_GFA[1])
build_baited_bloom_filter(ARGS)
POSITIVE_EXONS_BED = list(_find_exons_pipeline(ARGS))
SPLICE_GRAPH = build_splice_graph(POSITIVE_EXONS_BED)


def tearDownModule():
    """Remove temporary bloom and temporary GFA files"""
    # pylint: disable=invalid-name
    remove(TEMP_BLOOM[1])
    remove(TEMP_GFA[1])
    remove("/tmp/categories_multiMatch.fa")
    remove("/tmp/categories_noMatch.fa")
    remove("/tmp/categories_summary.tsv")
    remove("/tmp/categories_transcriptome.fa")



class TestPrepareSealer(TestCase, CustomAssertions):
    """_prepare_sealer(splice_graph, args)
    (nx.DiGraph, dict_of_parameters) -> str
    """

    def test_file_creation(self):
        """exfi.correct_splice_graph._prepare_sealer: test creation"""
        sealer_input_fn = _prepare_sealer(SPLICE_GRAPH, ARGS)
        actual = list(SeqIO.parse(sealer_input_fn, format="fasta"))
        expected = list(SeqIO.parse("tests/correct_splice_graph/to_seal.fa", format="fasta"))
        remove(sealer_input_fn)
        self.assertEqualListOfSeqrecords(actual, expected)


class TestRunSealer(TestCase, CustomAssertions):
    """_run_sealer(sealer_input_fn, args):
    (str, dict) -> str
    """

    def test_run(self):
{'fraction_non_alphanumeric': 0.10655571341587172, 'fraction_numerical': 0.006900750964075502, 'mean_word_length': 3.743022136669875, 'pattern_counts': {'":': 24, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, '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': '2345405', 'n_tokens_mistral': 1821, 'n_tokens_neox': 1749, 'n_words': 309}