<commit_before># -*- mode: python; coding: utf-8 -*-
# Copyright 2016 the HERA Collaboration
# Licensed under the 2-clause BSD license.

"""Testing for `hera_mc.connections`."""

import pytest
import numpy as np
from astropy.time import Time
from collections import OrderedDict

from hera_mc import cm_partconnect, cm_utils, cm_handling, cm_revisions, cm_dossier, cm_active


@pytest.fixture(scope='function')
def parts(mcsession):
    """Fixture for commonly used part data."""
    test_session = mcsession
    test_part = 'test_part'
    test_rev = 'Q'
    test_hptype = 'antenna'
    start_time = Time('2019-07-01 01:00:00', scale='utc')
    cm_handle = cm_handling.Handling(test_session)

    # Add a test part
    part = cm_partconnect.Parts()
    part.hpn = test_part
    part.hpn_rev = test_rev
    part.hptype = test_hptype
    part.manufacture_number = 'XYZ'
    part.start_gpstime = start_time.gps
    test_session.add(part)
    test_session.commit()

    class DataHolder(object):
        def __init__(self, test_session, test_part, test_rev, test_hptype,
                     start_time, cm_handle):
            self.test_session = test_session
            self.test_part = test_part
            self.test_rev = test_rev
            self.test_hptype = test_hptype
            self.start_time = start_time
            self.cm_handle = cm_handle

    parts = DataHolder(test_session, test_part, test_rev, test_hptype,
                       start_time, cm_handle)

    # yields the data we need but will continue to the del call after tests
    yield parts

    # some post-test object cleanup
    del(parts)

    return


def test_update_new(parts):
    ntp = 'new_test_part'
    data = [[ntp, 'X', 'hpn', ntp],
            [ntp, 'X', 'hpn_rev', 'X'],
            [ntp, 'X', 'hptype', 'antenna'],
            [ntp, 'X', 'start_gpstime', 1172530000]]
    cm_partconnect.update_part(parts.test_session, data)
    located = parts.cm_handle.get_dossier(hpn=[ntp], rev=['X'], at_date='now', exact_match=True)
    prkey = list(located.keys())[0]
    assert prkey == 'NEW_TEST_PART:X'


def test_find_part_type(parts):
    pt = parts.cm_handle.get_part_type_for(parts.test_part)
    assert pt == parts.test_hptype


def test_various_handling_utils(parts, capsys):
    parts.cm_handle._get_allowed_ports(['a'])
    assert parts.cm_handle.allowed_ports[0] == 'A'
    parts.cm_handle._get_allowed_ports('a,b')
    assert parts.cm_handle.allowed_ports[0] == 'A'
    parts.cm_handle._get_allowed_ports({"a": "A"})
    assert parts.cm_handle.allowed_ports is None


def test_apriori(mcsession):
    active = cm_active.ActiveData(mcsession)
    active.load_apriori()
    assert active.apriori['HH700:A'].status == 'not_connected'


def test_duplicate(mcsession):
    active = cm_active.ActiveData(mcsession)
    active.pytest_param = 'up'
    pytest.raises(ValueError, active.load_connections)
    active.pytest_param = 'down'
    pytest.raises(ValueError, active.load_connections)


def test_rosetta(mcsession, capsys):
    active = cm_active.ActiveData(mcsession)
    at_date = Time('2020-07-01 01:00:00', scale='utc')
    active.load_parts(at_date)
    active.load_rosetta(at_date)
    assert active.rosetta['SNPC000700'].syspn == 'heraNode700Snap700'
    print(active.rosetta['SNPC000700'])
    captured = capsys.readouterr()
    assert captured.out.strip().startswith('<SNPC000700')
    cm_partconnect.update_part_rosetta('SNPC000702', 'heraNode700Snap1',
                                       at_date, session=mcsession)
    active.load_rosetta(at_date)
    assert active.rosetta['SNPC000702'].syspn == 'heraNode700Snap2'
    stop_at = Time('2020-08-01 01:00:00', scale='utc')
    cm_partconnect.update_part_rosetta('SNPC000711', 'heraNode700Snap2',
                                       at_date, stop_at, mcsession)
    active.load_rosetta(Time('2020-07-15 01:00:00', scale='utc'))
    print(active.rosetta)
    assert int(active.rosetta['SNPC000711'].stop_gpstime) == 1280278818
    print("---",active.rosetta['SNPC000711'].stop_gpstime)
    cm_partconnect.update_part_rosetta('SNPC000709', 'heraNode700Snap709',
                                       stop_at, session=mcsession)
    assert int(active.rosetta['SNPC000709'].stop_gpstime) == 1280278818
    # Add a test part to fail on update part
    rose = cm_partconnect.PartRosetta()
    rose.hpn = 'SNPC000701'<commit_msg>fixed last test error<commit_after># -*- mode: python; coding: utf-8 -*-
# Copyright 2016 the HERA Collaboration
# Licensed under the 2-clause BSD license.

"""Testing for `hera_mc.connections`."""

import pytest
import numpy as np
from astropy.time import Time
from collections import OrderedDict

from hera_mc import cm_partconnect, cm_utils, cm_handling, cm_revisions, cm_dossier, cm_active


@pytest.fixture(scope='function')
def parts(mcsession):
    """Fixture for commonly used part data."""
    test_session = mcsession
    test_part = 'test_part'
    test_rev = 'Q'
    test_hptype = 'antenna'
    start_time = Time('2019-07-01 01:00:00', scale='utc')
    cm_handle = cm_handling.Handling(test_session)

    # Add a test part
    part = cm_partconnect.Parts()
    part.hpn = test_part
    part.hpn_rev = test_rev
    part.hptype = test_hptype
    part.manufacture_number = 'XYZ'
    part.start_gpstime = start_time.gps
    test_session.add(part)
    test_session.commit()

    class DataHolder(object):
        def __init__(self, test_session, test_part, test_rev, test_hptype,
                     start_time, cm_handle):
            self.test_session = test_session
            self.test_part = test_part
            self.test_rev = test_rev
            self.test_hptype = test_hptype
            self.start_time = start_time
            self.cm_handle = cm_handle

    parts = DataHolder(test_session, test_part, test_rev, test_hptype,
                       start_time, cm_handle)

    # yields the data we need but will continue to the del call after tests
    yield parts

    # some post-test object cleanup
    del(parts)

    return


def test_update_new(parts):
    ntp = 'new_test_part'
    data = [[ntp, 'X', 'hpn', ntp],
            [ntp, 'X', 'hpn_rev', 'X'],
            [ntp, 'X', 'hptype', 'antenna'],
            [ntp, 'X', 'start_gpstime', 1172530000]]
    cm_partconnect.update_part(parts.test_session, data)
    located = parts.cm_handle.get_dossier(hpn=[ntp], rev=['X'], at_date='now', exact_match=True)
    prkey = list(located.keys())[0]
    assert prkey == 'NEW_TEST_PART:X'


def test_find_part_type(parts):
    pt = parts.cm_handle.get_part_type_for(parts.test_part)
    assert pt == parts.test_hptype


def test_various_handling_utils(parts, capsys):
    parts.cm_handle._get_allowed_ports(['a'])
    assert parts.cm_handle.allowed_ports[0] == 'A'
    parts.cm_handle._get_allowed_ports('a,b')
    assert parts.cm_handle.allowed_ports[0] == 'A'
    parts.cm_handle._get_allowed_ports({"a": "A"})
    assert parts.cm_handle.allowed_ports is None


def test_apriori(mcsession):
    active = cm_active.ActiveData(mcsession)
    active.load_apriori()
    assert active.apriori['HH700:A'].status == 'not_connected'


def test_duplicate(mcsession):
    active = cm_active.ActiveData(mcsession)
    active.pytest_param = 'up'
    pytest.raises(ValueError, active.load_connections)
    active.pytest_param = 'down'
    pytest.raises(ValueError, active.load_connections)


def test_rosetta(mcsession, capsys):
    active = cm_active.ActiveData(mcsession)
    at_date = Time('2020-07-01 01:00:00', scale='utc')
    active.load_parts(at_date)
    active.load_rosetta(at_date)
    assert active.rosetta['SNPC000700'].syspn == 'heraNode700Snap700'
    print(active.rosetta['SNPC000700'])
    captured = capsys.readouterr()
    assert captured.out.strip().startswith('<SNPC000700')
    cm_partconnect.update_part_rosetta('SNPC000702', 'heraNode700Snap1',
                                       at_date, session=mcsession)
    active.load_rosetta(at_date)
    assert active.rosetta['SNPC000702'].syspn == 'heraNode700Snap2'
    stop_at = Time('2020-08-01 01:00:00', scale='utc')
    cm_partconnect.update_part_rosetta('SNPC000771', 'heraNode771Snap2',
                                       at_date, stop_at, session=mcsession)
    active.load_rosetta(Time('2020-07-15 01:00:00', scale='utc'))
    assert int(active.rosetta['SNPC000771'].stop_gpstime) == 1280278818
    cm_partconnect.update_part_rosetta('SNPC000709', 'heraNode700Snap709',
                                       stop_at, session=mcsession)
    assert int(active.rosetta['SNPC000709'].stop_gpstime) == 1280278818
    # Add a test part to fail on update part
    rose = cm_partconnect.PartRosetta()
    rose.hpn = 'SNPC000701'
{'fraction_non_alphanumeric': 0.11613199954007129, 'fraction_numerical': 0.042543405772105324, 'mean_word_length': 3.415228426395939, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3191509', 'n_tokens_mistral': 3303, 'n_tokens_neox': 3010, 'n_words': 550}