<commit_before>import unittest
import os
import psycopg2.extras
import datetime
import pytz

from pyiem.nws.products import parser
from pyiem.nws.products.vtec import parser as vtecparser
from pyiem.nws.ugc import UGC, UGCParseException
from pyiem.nws.nwsli import NWSLI
from pyiem.nws.products.lsr import LSRProductException


def get_file(name):
    ''' Helper function to get the text file contents '''
    basedir = os.path.dirname(__file__)
    fn = "%s/../../../data/product_examples/%s" % (basedir, name)
    return open(fn).read()


def utc(year, month, day, hour=0, minute=0):
    """UTC Timestamp generator"""
    return datetime.datetime(year, month, day, hour,
                             minute).replace(tzinfo=pytz.timezone("UTC"))


class TestProducts(unittest.TestCase):

    def setUp(self):
        ''' This is called for each test, beware '''
        self.dbconn = psycopg2.connect(database='postgis')
        self.txn = self.dbconn.cursor(cursor_factory=psycopg2.extras.DictCursor
                                      )

    def tearDown(self):
        ''' This is called after each test, beware '''
        self.dbconn.rollback()
        self.dbconn.close()

    def test_160513_windtag(self):
        """Wind tags can be in knots too!"""
        prod = vtecparser(get_file('SMWLWX.txt'))
        self.assertEquals(prod.segments[0].windtag, '34')
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0],
                          ("LWX issues Marine Warning [wind: &gt;34 KTS, "
                           "hail: 0.00 IN] for ((ANZ537)) [AN] till "
                           "May 13, 5:15 PM EDT "
                           "http://localhost#2016-O-NEW-KLWX-MA-W-0035"))

    def test_160415_mixedcase(self):
        """See how bad we break with mixed case"""
        prod = vtecparser(get_file('mIxEd_CaSe/FFSGLD.txt'))
        self.assertTrue(prod.tz is not None)
        self.assertEquals(prod.segments[0].vtec[0].action, 'CON')
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0],
                          ('GLD continues Flash Flood Warning for ((COC017)) '
                           '[CO] till Apr 15, 7:00 PM MDT '
                           'http://localhost#2016-O-CON-KGLD-FF-W-0001'))

    def test_151229_badgeo_lsr(self):
        """Make sure we reject a bad Geometry LSR"""
        utcnow = utc(2015, 12, 29, 18, 23)
        self.assertRaises(LSRProductException,
                          parser, get_file('LSRBOX.txt'), utcnow=utcnow)

    def test_151225_extfuture(self):
        """Warning failure jumps states!"""
        # /O.NEW.KPAH.FL.W.0093.151227T0517Z-151228T1727Z/
        prod = vtecparser(get_file('FLWPAH/FLWPAH_1.txt'))
        prod.sql(self.txn)
        self.txn.execute("""
            SELECT ugc, issue, expire from warnings_2015 where wfo = 'PAH'
            and phenomena = 'FL' and eventid = 93 and significance = 'W'
            and status = 'NEW'
        """)
        self.assertEquals(self.txn.rowcount, 2)
        # /O.EXT.KPAH.FL.W.0093.151227T0358Z-151229T0442Z/
        prod = vtecparser(get_file('FLWPAH/FLWPAH_2.txt'))
        prod.sql(self.txn)
        self.assertEquals(len(prod.warnings), 1, '\n'.join(prod.warnings))

    def test_150915_noexpire(self):
        """Check that we set an expiration for initial infinity SBW geo"""
        prod = vtecparser(get_file('FLWGRB.txt'))
        self.assertTrue(prod.segments[0].vtec[0].endts is None)
        prod.sql(self.txn)
        self.txn.execute("""
            SELECT init_expire, expire from sbw_2015 where wfo = 'GRB'
            and phenomena = 'FL' and eventid = 3 and significance = 'W'
            and status = 'NEW'
        """)
        row = self.txn.fetchone()
        self.assertTrue(row[0] is not None)
        self.assertTrue(row[1] is not None)

    def test_150820_exb(self):
        """Found a bug with setting of issuance for EXB case!"""
        for i in range(3):
            print('Parsing Product: %s.txt' % (i,))
            prod = vtecparser(get_file('CFWLWX/%i.txt' % (i,)))
            prod.sql(self.txn)
        # Make sure the issuance time is correct for MDZ014
        self.txn.execute("""SELECT issue at time zone 'UTC' from warnings_2015
        where wfo = 'LWX' and eventid = 30
        and phenomena = 'CF' and significance = 'Y'
        and ugc = 'MDZ014'""")
        self.assertEquals(self.txn.fetchone()[0],
                          datetime.datetime(2015, 8, 11, 13))

    def test_150814_init_expire(self):
        """ Make sure init_expire is not null"""
        prod = vtecparser(get_file('FLWLZK.txt'))
        prod.sql(self.txn)
        self.txn.execute("""SELECT count(*) from warnings_2015
            where wfo = 'LZK' and eventid = 18
            and phenomena = 'FL' and significance = 'W'
            and init_expire is null""")
        self.assertEquals(self.txn.fetchone()[0], 0)

    def test_150507_notcor(self):
        """SVROUN is not a product correction!"""
        prod = vtecparser(get_file('SVROUN.txt'))
        self.assertEquals(prod.is_correction(), False)

    def test_150429_flswithsign(self):
        """FLSMKX see that we are okay with the signature"""
        prod = vtecparser(get_file('FLSMKX.txt'))
        self.assertEqual(prod.segments[0].giswkt,
                         ('SRID=4326;MULTIPOLYGON '
                          '(((-88.320000 42.620000, -88.130000 42.620000, '
                          '-88.120000 42.520000, -88.100000 42.450000, '
                          '-88.270000 42.450000, -88.250000 42.550000, '
                          '-88.320000 42.620000)))'))

    def test_150422_tornadomag(self):
        """LSRTAE see what we do with tornado magitnudes"""
        utcnow = utc(2015, 4, 22, 15, 20)
        prod = parser(get_file('LSRTAE.txt'), utcnow=utcnow)
        j = prod.get_jabbers('http://iem.local/')
        self.assertEqual(j[0][1], (
            '<p>4 W Bruce [Walton Co, FL] NWS EMPLOYEE '
            '<a href="http://iem.local/#TAE/201504191322/201504191322">'
            'reports TORNADO of EF0</a> at 19 Apr, 9:22 AM EDT -- '
            'SHORT EF0 TORNADO PATH CONFIRMED BY NWS DUAL POL RADAR DEBRIS '
            'SIGNATURE IN A RURAL AREA WEST OF BRUCE. DAMAGE LIKELY CONFINED '
            'TO TREES. ESTIMATED DURATION 3 MINUTES. PATH LENGTH '
            'APPROXIMATELY 1 MILE.</p>'))
        self.assertEqual(j[0][2]['twitter'], (
            'At 9:22 AM, 4 W Bruce [Walton Co, FL] NWS EMPLOYEE reports '
            'TORNADO of EF0 #TAE '
            'http://iem.local/#TAE/201504191322/201504191322'))

    def test_150331_notcorrection(self):
        """SVRMEG is not a product correction"""
        prod = vtecparser(get_file('SVRMEG.txt'))
        self.assertTrue(not prod.is_correction())

    def test_150304_testtor(self):
        """TORILX is a test, we had better handle it!"""
        prod = vtecparser(get_file('TORILX.txt'))
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(len(j), 0)

    def test_150203_exp_does_not_end(self):
        """MWWCAR a VTEC EXP action should not terminate it """
        for i in range(24):
            print('Parsing Product: %s.txt' % (i,))
            prod = vtecparser(get_file('MWWCAR/%i.txt' % (i,)))
            prod.sql(self.txn)
            self.assertEquals(len(prod.warnings), 0)

    def test_150203_null_issue(self):
        """WSWOKX had null issue times, bad! """
        for i in range(18):
            print('Parsing Product: %s.txt' % (i,))
            prod = vtecparser(get_file('WSWOKX/%i.txt' % (i,)))
            prod.sql(self.txn)
            # Make sure there are no null issue times
            self.txn.execute("""SELECT count(*) from warnings_2015
            where wfo = 'OKX' and eventid = 6
            and phenomena = 'WW' and significance = 'Y'
            and issue is null""")
            self.assertEquals(self.txn.fetchone()[0], 0)

    def test_150202_hwo(self):
        """HWORNK emitted a poorly worded error message"""
        prod = parser(get_file('HWORNK.txt'))
        self.assertRaises(Exception, prod.get_jabbers,
                          'http://localhost', 'http://localhost')

    def test_160418_hwospn(self):
        """Make sure a spanish HWO does not trip us up..."""
        prod = parser(get_file('HWOSPN.txt'))
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0],
                          ("JSJ issues Hazardous Weather Outlook (HWO) "
                           "http://localhost?"
                           "pid=201604181018-TJSJ-FLCA42-HWOSPN"))

    def test_150115_correction_sbw(self):
        """ FLWMHX make sure a correction does not result in two polygons """
        prod = vtecparser(get_file('FLWMHX/0.txt'))
        prod.sql(self.txn)
        self.assertEquals(len(prod.warnings), 0, "\n".join(prod.warnings))
        prod = vtecparser(get_file('FLWMHX/1.txt'))
        prod.sql(self.txn)
        self.assertEquals(len(prod.warnings), 0, "\n".join(prod.warnings))

    def test_150105_considerable_tag(self):
        """ TORFSD has considerable tag """
        prod = vtecparser(get_file('TORFSD.txt'))
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0], (
            'FSD issues Tornado Warning '
            '[tornado: RADAR INDICATED, tornado damage threat: CONSIDERABLE, '
            'hail: 1.50 IN] for ((IAC035)) [IA] till 8:00 PM CDT * AT 720 '
            'PM CDT...A SEVERE THUNDERSTORM CAPABLE OF PRODUCING A LARGE '
            'AND EXTREMELY DANGEROUS TORNADO WAS LOCATED NEAR WASHTA...AND '
            'MOVING NORTHEAST AT 30 MPH. '
            'http://localhost#2013-O-NEW-KFSD-TO-W-0020'))

    def test_150105_sbw(self):
        """ FLSLBF SBW that spans two years! """
        for i in range(7):<commit_msg>remove ambigity with unittest<commit_after>import unittest
import os
import psycopg2.extras
import datetime
import pytz

from pyiem.nws.products import parser
from pyiem.nws.products.vtec import parser as vtecparser
from pyiem.nws.ugc import UGC, UGCParseException
from pyiem.nws.nwsli import NWSLI
from pyiem.nws.products.lsr import LSRProductException


def get_file(name):
    ''' Helper function to get the text file contents '''
    basedir = os.path.dirname(__file__)
    fn = "%s/../../../data/product_examples/%s" % (basedir, name)
    return open(fn).read()


def utc(year, month, day, hour=0, minute=0):
    """UTC Timestamp generator"""
    return datetime.datetime(year, month, day, hour,
                             minute).replace(tzinfo=pytz.timezone("UTC"))


class TestProducts(unittest.TestCase):

    def setUp(self):
        ''' This is called for each test, beware '''
        self.dbconn = psycopg2.connect(database='postgis')
        self.txn = self.dbconn.cursor(cursor_factory=psycopg2.extras.DictCursor
                                      )

    def tearDown(self):
        ''' This is called after each test, beware '''
        self.dbconn.rollback()
        self.dbconn.close()

    def test_160513_windtag(self):
        """Wind tags can be in knots too!"""
        prod = vtecparser(get_file('SMWLWX.txt'))
        self.assertEquals(prod.segments[0].windtag, '34')
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0],
                          ("LWX issues Marine Warning [wind: &gt;34 KTS, "
                           "hail: 0.00 IN] for ((ANZ537)) [AN] till "
                           "May 13, 5:15 PM EDT "
                           "http://localhost#2016-O-NEW-KLWX-MA-W-0035"))

    def test_160415_mixedcase(self):
        """See how bad we break with mixed case"""
        prod = vtecparser(get_file('mIxEd_CaSe/FFSGLD.txt'))
        self.assertTrue(prod.tz is not None)
        self.assertEquals(prod.segments[0].vtec[0].action, 'CON')
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0],
                          ('GLD continues Flash Flood Warning for ((COC017)) '
                           '[CO] till Apr 15, 7:00 PM MDT '
                           'http://localhost#2016-O-CON-KGLD-FF-W-0001'))

    def test_151229_badgeo_lsr(self):
        """Make sure we reject a bad Geometry LSR"""
        utcnow = utc(2015, 12, 29, 18, 23)
        self.assertRaises(LSRProductException,
                          parser, get_file('LSRBOX.txt'), utcnow=utcnow)

    def test_151225_extfuture(self):
        """Warning failure jumps states!"""
        # /O.NEW.KPAH.FL.W.0093.151227T0517Z-151228T1727Z/
        prod = vtecparser(get_file('FLWPAH/FLWPAH_1.txt'))
        prod.sql(self.txn)
        self.txn.execute("""
            SELECT ugc, issue, expire from warnings_2015 where wfo = 'PAH'
            and phenomena = 'FL' and eventid = 93 and significance = 'W'
            and status = 'NEW'
        """)
        self.assertEquals(self.txn.rowcount, 2)
        # /O.EXT.KPAH.FL.W.0093.151227T0358Z-151229T0442Z/
        prod = vtecparser(get_file('FLWPAH/FLWPAH_2.txt'))
        prod.sql(self.txn)
        self.assertEquals(len(prod.warnings), 1, '\n'.join(prod.warnings))

    def test_150915_noexpire(self):
        """Check that we set an expiration for initial infinity SBW geo"""
        prod = vtecparser(get_file('FLWGRB.txt'))
        self.assertTrue(prod.segments[0].vtec[0].endts is None)
        prod.sql(self.txn)
        self.txn.execute("""
            SELECT init_expire, expire from sbw_2015 where wfo = 'GRB'
            and phenomena = 'FL' and eventid = 3 and significance = 'W'
            and status = 'NEW'
        """)
        row = self.txn.fetchone()
        self.assertTrue(row[0] is not None)
        self.assertTrue(row[1] is not None)

    def test_150820_exb(self):
        """Found a bug with setting of issuance for EXB case!"""
        for i in range(3):
            print('Parsing Product: %s.txt' % (i,))
            prod = vtecparser(get_file('CFWLWX/%i.txt' % (i,)))
            prod.sql(self.txn)
        # Make sure the issuance time is correct for MDZ014
        self.txn.execute("""SELECT issue at time zone 'UTC' from warnings_2015
        where wfo = 'LWX' and eventid = 30
        and phenomena = 'CF' and significance = 'Y'
        and ugc = 'MDZ014'""")
        self.assertEquals(self.txn.fetchone()[0],
                          datetime.datetime(2015, 8, 11, 13))

    def test_150814_init_expire(self):
        """ Make sure init_expire is not null"""
        prod = vtecparser(get_file('FLWLZK.txt'))
        prod.sql(self.txn)
        self.txn.execute("""SELECT count(*) from warnings_2015
            where wfo = 'LZK' and eventid = 18
            and phenomena = 'FL' and significance = 'W'
            and init_expire is null""")
        self.assertEquals(self.txn.fetchone()[0], 0)

    def test_150507_notcor(self):
        """SVROUN is not a product correction!"""
        prod = vtecparser(get_file('SVROUN.txt'))
        self.assertEquals(prod.is_correction(), False)

    def test_150429_flswithsign(self):
        """FLSMKX see that we are okay with the signature"""
        prod = vtecparser(get_file('FLSMKX.txt'))
        self.assertEqual(prod.segments[0].giswkt,
                         ('SRID=4326;MULTIPOLYGON '
                          '(((-88.320000 42.620000, -88.130000 42.620000, '
                          '-88.120000 42.520000, -88.100000 42.450000, '
                          '-88.270000 42.450000, -88.250000 42.550000, '
                          '-88.320000 42.620000)))'))

    def test_150422_tornadomag(self):
        """LSRTAE see what we do with tornado magitnudes"""
        utcnow = utc(2015, 4, 22, 15, 20)
        prod = parser(get_file('LSRTAE.txt'), utcnow=utcnow)
        j = prod.get_jabbers('http://iem.local/')
        self.assertEqual(j[0][1], (
            '<p>4 W Bruce [Walton Co, FL] NWS EMPLOYEE '
            '<a href="http://iem.local/#TAE/201504191322/201504191322">'
            'reports TORNADO of EF0</a> at 19 Apr, 9:22 AM EDT -- '
            'SHORT EF0 TORNADO PATH CONFIRMED BY NWS DUAL POL RADAR DEBRIS '
            'SIGNATURE IN A RURAL AREA WEST OF BRUCE. DAMAGE LIKELY CONFINED '
            'TO TREES. ESTIMATED DURATION 3 MINUTES. PATH LENGTH '
            'APPROXIMATELY 1 MILE.</p>'))
        self.assertEqual(j[0][2]['twitter'], (
            'At 9:22 AM, 4 W Bruce [Walton Co, FL] NWS EMPLOYEE reports '
            'TORNADO of EF0 #TAE '
            'http://iem.local/#TAE/201504191322/201504191322'))

    def test_150331_notcorrection(self):
        """SVRMEG is not a product correction"""
        prod = vtecparser(get_file('SVRMEG.txt'))
        self.assertTrue(not prod.is_correction())

    def test_150304_testtor(self):
        """TORILX is a test, we had better handle it!"""
        prod = vtecparser(get_file('TORILX.txt'))
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(len(j), 0)

    def test_150203_exp_does_not_end(self):
        """MWWCAR a VTEC EXP action should not terminate it """
        for i in range(24):
            print('Parsing Product: %s.txt' % (i,))
            prod = vtecparser(get_file('MWWCAR/%i.txt' % (i,)))
            prod.sql(self.txn)
            self.assertEquals(len(prod.warnings), 0)

    def test_150203_null_issue(self):
        """WSWOKX had null issue times, bad! """
        for i in range(18):
            print('Parsing Product: %s.txt' % (i,))
            prod = vtecparser(get_file('WSWOKX/%i.txt' % (i,)))
            prod.sql(self.txn)
            # Make sure there are no null issue times
            self.txn.execute("""SELECT count(*) from warnings_2015
            where wfo = 'OKX' and eventid = 6
            and phenomena = 'WW' and significance = 'Y'
            and issue is null""")
            self.assertEquals(self.txn.fetchone()[0], 0)

    def test_150202_hwo(self):
        """HWORNK emitted a poorly worded error message"""
        prod = parser(get_file('HWORNK.txt'))
        self.assertRaises(Exception, prod.get_jabbers,
                          'http://localhost', 'http://localhost')

    def test_160418_hwospn(self):
        """Make sure a spanish HWO does not trip us up..."""
        utcnow = utc(2016, 4, 18, 10, 10)
        prod = parser(get_file('HWOSPN.txt'), utcnow=utcnow)
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0],
                          ("JSJ issues Hazardous Weather Outlook (HWO) "
                           "http://localhost?"
                           "pid=201604181018-TJSJ-FLCA42-HWOSPN"))

    def test_150115_correction_sbw(self):
        """ FLWMHX make sure a correction does not result in two polygons """
        prod = vtecparser(get_file('FLWMHX/0.txt'))
        prod.sql(self.txn)
        self.assertEquals(len(prod.warnings), 0, "\n".join(prod.warnings))
        prod = vtecparser(get_file('FLWMHX/1.txt'))
        prod.sql(self.txn)
        self.assertEquals(len(prod.warnings), 0, "\n".join(prod.warnings))

    def test_150105_considerable_tag(self):
        """ TORFSD has considerable tag """
        prod = vtecparser(get_file('TORFSD.txt'))
        j = prod.get_jabbers('http://localhost', 'http://localhost')
        self.assertEquals(j[0][0], (
            'FSD issues Tornado Warning '
            '[tornado: RADAR INDICATED, tornado damage threat: CONSIDERABLE, '
            'hail: 1.50 IN] for ((IAC035)) [IA] till 8:00 PM CDT * AT 720 '
            'PM CDT...A SEVERE THUNDERSTORM CAPABLE OF PRODUCING A LARGE '
            'AND EXTREMELY DANGEROUS TORNADO WAS LOCATED NEAR WASHTA...AND '
            'MOVING NORTHEAST AT 30 MPH. '
            'http://localhost#2013-O-NEW-KFSD-TO-W-0020'))

    def test_150105_sbw(self):
        """ FLSLBF SBW that spans two years! """
        for i in range(7):
{'fraction_non_alphanumeric': 0.13282856139998997, 'fraction_numerical': 0.053502482073910645, 'mean_word_length': 2.2002567394094994, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1342666', 'n_tokens_mistral': 7453, 'n_tokens_neox': 6572, 'n_words': 1501}