diff --git a/Makefile b/Makefile index 5340686..e6cb2aa 100644 --- a/Makefile +++ b/Makefile @@ -55,4 +55,4 @@ check: .venv/bin/tox # Code format check with isort and black tox -epep8 fix: .venv/bin/tox # fixes code formatting with isort and black - tox -efmt \ No newline at end of file + tox -efmt diff --git a/aprsd/plugin.py b/aprsd/plugin.py index eae9569..224b475 100644 --- a/aprsd/plugin.py +++ b/aprsd/plugin.py @@ -3,19 +3,11 @@ import abc import fnmatch import importlib import inspect -import json import logging import os import re -import shutil -import subprocess -import time -import aprsd -from aprsd import email, messaging -from aprsd.fuzzyclock import fuzzy import pluggy -import requests from thesmuggler import smuggle # setup the global logger @@ -25,17 +17,61 @@ hookspec = pluggy.HookspecMarker("aprsd") hookimpl = pluggy.HookimplMarker("aprsd") CORE_PLUGINS = [ - "aprsd.plugin.EmailPlugin", - "aprsd.plugin.FortunePlugin", - "aprsd.plugin.LocationPlugin", - "aprsd.plugin.PingPlugin", - "aprsd.plugin.QueryPlugin", - "aprsd.plugin.TimePlugin", - "aprsd.plugin.WeatherPlugin", - "aprsd.plugin.VersionPlugin", + "aprsd.plugins.email.EmailPlugin", + "aprsd.plugins.fortune.FortunePlugin", + "aprsd.plugins.location.LocationPlugin", + "aprsd.plugins.ping.PingPlugin", + "aprsd.plugins.query.QueryPlugin", + "aprsd.plugins.time.TimePlugin", + "aprsd.plugins.weather.WeatherPlugin", + "aprsd.plugins.version.VersionPlugin", ] +class APRSDCommandSpec: + """A hook specification namespace.""" + + @hookspec + def run(self, fromcall, message, ack): + """My special little hook that you can customize.""" + pass + + +class APRSDPluginBase(metaclass=abc.ABCMeta): + def __init__(self, config): + """The aprsd config object is stored.""" + self.config = config + + @property + def command_name(self): + """The usage string help.""" + raise NotImplementedError + + @property + def command_regex(self): + """The regex to match from the caller""" + raise NotImplementedError + + @property + def version(self): + """Version""" + raise NotImplementedError + + @hookimpl + def run(self, fromcall, message, ack): + if re.search(self.command_regex, message): + return self.command(fromcall, message, ack) + + @abc.abstractmethod + def command(self, fromcall, message, ack): + """This is the command that runs when the regex matches. + + To reply with a message over the air, return a string + to send. + """ + pass + + class PluginManager: # The singleton instance object for this class _instance = None @@ -197,366 +233,3 @@ class PluginManager: def get_plugins(self): return self._pluggy_pm.get_plugins() - - -class APRSDCommandSpec: - """A hook specification namespace.""" - - @hookspec - def run(self, fromcall, message, ack): - """My special little hook that you can customize.""" - pass - - -class APRSDPluginBase(metaclass=abc.ABCMeta): - def __init__(self, config): - """The aprsd config object is stored.""" - self.config = config - - @property - def command_name(self): - """The usage string help.""" - raise NotImplementedError - - @property - def command_regex(self): - """The regex to match from the caller""" - raise NotImplementedError - - @property - def version(self): - """Version""" - raise NotImplementedError - - @hookimpl - def run(self, fromcall, message, ack): - if re.search(self.command_regex, message): - return self.command(fromcall, message, ack) - - @abc.abstractmethod - def command(self, fromcall, message, ack): - """This is the command that runs when the regex matches. - - To reply with a message over the air, return a string - to send. - """ - pass - - -class FortunePlugin(APRSDPluginBase): - """Fortune.""" - - version = "1.0" - command_regex = "^[fF]" - command_name = "fortune" - - def command(self, fromcall, message, ack): - LOG.info("FortunePlugin") - reply = None - - fortune_path = shutil.which("fortune") - if not fortune_path: - reply = "Fortune command not installed" - return reply - - try: - process = subprocess.Popen( - [fortune_path, "-s", "-n 60"], - stdout=subprocess.PIPE, - ) - reply = process.communicate()[0] - reply = reply.decode(errors="ignore").rstrip() - except Exception as ex: - reply = "Fortune command failed '{}'".format(ex) - LOG.error(reply) - - return reply - - -class LocationPlugin(APRSDPluginBase): - """Location!""" - - version = "1.0" - command_regex = "^[lL]" - command_name = "location" - - config_items = {"apikey": "aprs.fi api key here"} - - def command(self, fromcall, message, ack): - LOG.info("Location Plugin") - # get last location of a callsign, get descriptive name from weather service - try: - # optional second argument is a callsign to search - a = re.search(r"^.*\s+(.*)", message) - if a is not None: - searchcall = a.group(1) - searchcall = searchcall.upper() - else: - # if no second argument, search for calling station - searchcall = fromcall - url = ( - "http://api.aprs.fi/api/get?name=" - + searchcall - + "&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json" - ) - response = requests.get(url) - # aprs_data = json.loads(response.read()) - aprs_data = json.loads(response.text) - LOG.debug("LocationPlugin: aprs_data = {}".format(aprs_data)) - lat = aprs_data["entries"][0]["lat"] - lon = aprs_data["entries"][0]["lng"] - try: # altitude not always provided - alt = aprs_data["entries"][0]["altitude"] - except Exception: - alt = 0 - altfeet = int(alt * 3.28084) - aprs_lasttime_seconds = aprs_data["entries"][0]["lasttime"] - # aprs_lasttime_seconds = aprs_lasttime_seconds.encode( - # "ascii", errors="ignore" - # ) # unicode to ascii - delta_seconds = time.time() - int(aprs_lasttime_seconds) - delta_hours = delta_seconds / 60 / 60 - url2 = ( - "https://forecast.weather.gov/MapClick.php?lat=" - + str(lat) - + "&lon=" - + str(lon) - + "&FcstType=json" - ) - response2 = requests.get(url2) - wx_data = json.loads(response2.text) - - reply = "{}: {} {}' {},{} {}h ago".format( - searchcall, - wx_data["location"]["areaDescription"], - str(altfeet), - str(alt), - str(lon), - str("%.1f" % round(delta_hours, 1)), - ).rstrip() - except Exception as e: - LOG.debug("Locate failed with: " + "%s" % str(e)) - reply = "Unable to find station " + searchcall + ". Sending beacons?" - - return reply - - -class PingPlugin(APRSDPluginBase): - """Ping.""" - - version = "1.0" - command_regex = "^[pP]" - command_name = "ping" - - def command(self, fromcall, message, ack): - LOG.info("PINGPlugin") - stm = time.localtime() - h = stm.tm_hour - m = stm.tm_min - s = stm.tm_sec - reply = ( - "Pong! " + str(h).zfill(2) + ":" + str(m).zfill(2) + ":" + str(s).zfill(2) - ) - return reply.rstrip() - - -class QueryPlugin(APRSDPluginBase): - """Query command.""" - - version = "1.0" - command_regex = r"^\?.*" - command_name = "query" - - def command(self, fromcall, message, ack): - LOG.info("Query COMMAND") - - tracker = messaging.MsgTrack() - reply = "Pending Messages ({})".format(len(tracker)) - - searchstring = "^" + self.config["ham"]["callsign"] + ".*" - # only I can do admin commands - if re.search(searchstring, fromcall): - r = re.search(r"^\?-\*", message) - if r is not None: - if len(tracker) > 0: - reply = "Resend ALL Delayed msgs" - LOG.debug(reply) - tracker.restart_delayed() - else: - reply = "No Delayed Msgs" - LOG.debug(reply) - return reply - - r = re.search(r"^\?-[fF]!", message) - if r is not None: - reply = "Deleting ALL Delayed msgs." - LOG.debug(reply) - tracker.flush() - return reply - - return reply - - -class TimePlugin(APRSDPluginBase): - """Time command.""" - - version = "1.0" - command_regex = "^[tT]" - command_name = "time" - - def command(self, fromcall, message, ack): - LOG.info("TIME COMMAND") - stm = time.localtime() - h = stm.tm_hour - m = stm.tm_min - cur_time = fuzzy(h, m, 1) - reply = "{} ({}:{} PDT) ({})".format( - cur_time, - str(h), - str(m).rjust(2, "0"), - message.rstrip(), - ) - return reply - - -class WeatherPlugin(APRSDPluginBase): - """Weather Command""" - - version = "1.0" - command_regex = "^[wW]" - command_name = "weather" - - def command(self, fromcall, message, ack): - LOG.info("Weather Plugin") - try: - url = ( - "http://api.aprs.fi/api/get?" - "&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json" - "&name=%s" % fromcall - ) - response = requests.get(url) - # aprs_data = json.loads(response.read()) - aprs_data = json.loads(response.text) - lat = aprs_data["entries"][0]["lat"] - lon = aprs_data["entries"][0]["lng"] - url2 = ( - "https://forecast.weather.gov/MapClick.php?lat=%s" - "&lon=%s&FcstType=json" % (lat, lon) - ) - response2 = requests.get(url2) - # wx_data = json.loads(response2.read()) - wx_data = json.loads(response2.text) - reply = ( - "%sF(%sF/%sF) %s. %s, %s." - % ( - wx_data["currentobservation"]["Temp"], - wx_data["data"]["temperature"][0], - wx_data["data"]["temperature"][1], - wx_data["data"]["weather"][0], - wx_data["time"]["startPeriodName"][1], - wx_data["data"]["weather"][1], - ) - ).rstrip() - LOG.debug("reply: '{}' ".format(reply)) - except Exception as e: - LOG.debug("Weather failed with: " + "%s" % str(e)) - reply = "Unable to find you (send beacon?)" - - return reply - - -class EmailPlugin(APRSDPluginBase): - """Email Plugin.""" - - version = "1.0" - command_regex = "^-.*" - command_name = "email" - - # message_number:time combos so we don't resend the same email in - # five mins {int:int} - email_sent_dict = {} - - def command(self, fromcall, message, ack): - LOG.info("Email COMMAND") - reply = None - - searchstring = "^" + self.config["ham"]["callsign"] + ".*" - # only I can do email - if re.search(searchstring, fromcall): - # digits only, first one is number of emails to resend - r = re.search("^-([0-9])[0-9]*$", message) - if r is not None: - LOG.debug("RESEND EMAIL") - email.resend_email(r.group(1), fromcall) - reply = messaging.NULL_MESSAGE - # -user@address.com body of email - elif re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message): - # (same search again) - a = re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message) - if a is not None: - to_addr = a.group(1) - content = a.group(2) - - email_address = email.get_email_from_shortcut(to_addr) - if not email_address: - reply = "Bad email address" - return reply - - # send recipient link to aprs.fi map - if content == "mapme": - content = "Click for my location: http://aprs.fi/{}".format( - self.config["ham"]["callsign"], - ) - too_soon = 0 - now = time.time() - # see if we sent this msg number recently - if ack in self.email_sent_dict: - # BUG(hemna) - when we get a 2 different email command - # with the same ack #, we don't send it. - timedelta = now - self.email_sent_dict[ack] - if timedelta < 300: # five minutes - too_soon = 1 - if not too_soon or ack == 0: - LOG.info("Send email '{}'".format(content)) - send_result = email.send_email(to_addr, content) - reply = messaging.NULL_MESSAGE - if send_result != 0: - reply = "-{} failed".format(to_addr) - # messaging.send_message(fromcall, "-" + to_addr + " failed") - else: - # clear email sent dictionary if somehow goes over 100 - if len(self.email_sent_dict) > 98: - LOG.debug( - "DEBUG: email_sent_dict is big (" - + str(len(self.email_sent_dict)) - + ") clearing out.", - ) - self.email_sent_dict.clear() - self.email_sent_dict[ack] = now - else: - LOG.info( - "Email for message number " - + ack - + " recently sent, not sending again.", - ) - else: - reply = "Bad email address" - # messaging.send_message(fromcall, "Bad email address") - - return reply - - -class VersionPlugin(APRSDPluginBase): - """Version of APRSD Plugin.""" - - version = "1.0" - command_regex = "^[vV]" - command_name = "version" - - # message_number:time combos so we don't resend the same email in - # five mins {int:int} - email_sent_dict = {} - - def command(self, fromcall, message, ack): - LOG.info("Version COMMAND") - return "APRSD version '{}'".format(aprsd.__version__) diff --git a/aprsd/plugins/__init__.py b/aprsd/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/aprsd/plugins/email.py b/aprsd/plugins/email.py new file mode 100644 index 0000000..1372327 --- /dev/null +++ b/aprsd/plugins/email.py @@ -0,0 +1,88 @@ +import logging +import re +import time + +from aprsd import email, messaging, plugin + +LOG = logging.getLogger("APRSD") + + +class EmailPlugin(plugin.APRSDPluginBase): + """Email Plugin.""" + + version = "1.0" + command_regex = "^-.*" + command_name = "email" + + # message_number:time combos so we don't resend the same email in + # five mins {int:int} + email_sent_dict = {} + + def command(self, fromcall, message, ack): + LOG.info("Email COMMAND") + reply = None + + searchstring = "^" + self.config["ham"]["callsign"] + ".*" + # only I can do email + if re.search(searchstring, fromcall): + # digits only, first one is number of emails to resend + r = re.search("^-([0-9])[0-9]*$", message) + if r is not None: + LOG.debug("RESEND EMAIL") + email.resend_email(r.group(1), fromcall) + reply = messaging.NULL_MESSAGE + # -user@address.com body of email + elif re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message): + # (same search again) + a = re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message) + if a is not None: + to_addr = a.group(1) + content = a.group(2) + + email_address = email.get_email_from_shortcut(to_addr) + if not email_address: + reply = "Bad email address" + return reply + + # send recipient link to aprs.fi map + if content == "mapme": + content = "Click for my location: http://aprs.fi/{}".format( + self.config["ham"]["callsign"], + ) + too_soon = 0 + now = time.time() + # see if we sent this msg number recently + if ack in self.email_sent_dict: + # BUG(hemna) - when we get a 2 different email command + # with the same ack #, we don't send it. + timedelta = now - self.email_sent_dict[ack] + if timedelta < 300: # five minutes + too_soon = 1 + if not too_soon or ack == 0: + LOG.info("Send email '{}'".format(content)) + send_result = email.send_email(to_addr, content) + reply = messaging.NULL_MESSAGE + if send_result != 0: + reply = "-{} failed".format(to_addr) + # messaging.send_message(fromcall, "-" + to_addr + " failed") + else: + # clear email sent dictionary if somehow goes over 100 + if len(self.email_sent_dict) > 98: + LOG.debug( + "DEBUG: email_sent_dict is big (" + + str(len(self.email_sent_dict)) + + ") clearing out.", + ) + self.email_sent_dict.clear() + self.email_sent_dict[ack] = now + else: + LOG.info( + "Email for message number " + + ack + + " recently sent, not sending again.", + ) + else: + reply = "Bad email address" + # messaging.send_message(fromcall, "Bad email address") + + return reply diff --git a/aprsd/plugins/fortune.py b/aprsd/plugins/fortune.py new file mode 100644 index 0000000..52e59e1 --- /dev/null +++ b/aprsd/plugins/fortune.py @@ -0,0 +1,37 @@ +import logging +import shutil +import subprocess + +from aprsd import plugin + +LOG = logging.getLogger("APRSD") + + +class FortunePlugin(plugin.APRSDPluginBase): + """Fortune.""" + + version = "1.0" + command_regex = "^[fF]" + command_name = "fortune" + + def command(self, fromcall, message, ack): + LOG.info("FortunePlugin") + reply = None + + fortune_path = shutil.which("fortune") + if not fortune_path: + reply = "Fortune command not installed" + return reply + + try: + process = subprocess.Popen( + [fortune_path, "-s", "-n 60"], + stdout=subprocess.PIPE, + ) + reply = process.communicate()[0] + reply = reply.decode(errors="ignore").rstrip() + except Exception as ex: + reply = "Fortune command failed '{}'".format(ex) + LOG.error(reply) + + return reply diff --git a/aprsd/plugins/location.py b/aprsd/plugins/location.py new file mode 100644 index 0000000..be75e34 --- /dev/null +++ b/aprsd/plugins/location.py @@ -0,0 +1,77 @@ +import json +import logging +import re +import time + +from aprsd import plugin +import requests + +LOG = logging.getLogger("APRSD") + + +class LocationPlugin(plugin.APRSDPluginBase): + """Location!""" + + version = "1.0" + command_regex = "^[lL]" + command_name = "location" + + config_items = {"apikey": "aprs.fi api key here"} + + def command(self, fromcall, message, ack): + LOG.info("Location Plugin") + # get last location of a callsign, get descriptive name from weather service + try: + # optional second argument is a callsign to search + a = re.search(r"^.*\s+(.*)", message) + if a is not None: + searchcall = a.group(1) + searchcall = searchcall.upper() + else: + # if no second argument, search for calling station + searchcall = fromcall + url = ( + "http://api.aprs.fi/api/get?name=" + + searchcall + + "&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json" + ) + response = requests.get(url) + # aprs_data = json.loads(response.read()) + aprs_data = json.loads(response.text) + LOG.debug("LocationPlugin: aprs_data = {}".format(aprs_data)) + lat = aprs_data["entries"][0]["lat"] + lon = aprs_data["entries"][0]["lng"] + try: # altitude not always provided + alt = aprs_data["entries"][0]["altitude"] + except Exception: + alt = 0 + altfeet = int(alt * 3.28084) + aprs_lasttime_seconds = aprs_data["entries"][0]["lasttime"] + # aprs_lasttime_seconds = aprs_lasttime_seconds.encode( + # "ascii", errors="ignore" + # ) # unicode to ascii + delta_seconds = time.time() - int(aprs_lasttime_seconds) + delta_hours = delta_seconds / 60 / 60 + url2 = ( + "https://forecast.weather.gov/MapClick.php?lat=" + + str(lat) + + "&lon=" + + str(lon) + + "&FcstType=json" + ) + response2 = requests.get(url2) + wx_data = json.loads(response2.text) + + reply = "{}: {} {}' {},{} {}h ago".format( + searchcall, + wx_data["location"]["areaDescription"], + str(altfeet), + str(alt), + str(lon), + str("%.1f" % round(delta_hours, 1)), + ).rstrip() + except Exception as e: + LOG.debug("Locate failed with: " + "%s" % str(e)) + reply = "Unable to find station " + searchcall + ". Sending beacons?" + + return reply diff --git a/aprsd/plugins/ping.py b/aprsd/plugins/ping.py new file mode 100644 index 0000000..754da05 --- /dev/null +++ b/aprsd/plugins/ping.py @@ -0,0 +1,25 @@ +import logging +import time + +from aprsd import plugin + +LOG = logging.getLogger("APRSD") + + +class PingPlugin(plugin.APRSDPluginBase): + """Ping.""" + + version = "1.0" + command_regex = "^[pP]" + command_name = "ping" + + def command(self, fromcall, message, ack): + LOG.info("PINGPlugin") + stm = time.localtime() + h = stm.tm_hour + m = stm.tm_min + s = stm.tm_sec + reply = ( + "Pong! " + str(h).zfill(2) + ":" + str(m).zfill(2) + ":" + str(s).zfill(2) + ) + return reply.rstrip() diff --git a/aprsd/plugins/query.py b/aprsd/plugins/query.py new file mode 100644 index 0000000..e5a4bfc --- /dev/null +++ b/aprsd/plugins/query.py @@ -0,0 +1,43 @@ +import logging +import re + +from aprsd import messaging, plugin + +LOG = logging.getLogger("APRSD") + + +class QueryPlugin(plugin.APRSDPluginBase): + """Query command.""" + + version = "1.0" + command_regex = r"^\?.*" + command_name = "query" + + def command(self, fromcall, message, ack): + LOG.info("Query COMMAND") + + tracker = messaging.MsgTrack() + reply = "Pending Messages ({})".format(len(tracker)) + + searchstring = "^" + self.config["ham"]["callsign"] + ".*" + # only I can do admin commands + if re.search(searchstring, fromcall): + r = re.search(r"^\?-\*", message) + if r is not None: + if len(tracker) > 0: + reply = "Resend ALL Delayed msgs" + LOG.debug(reply) + tracker.restart_delayed() + else: + reply = "No Delayed Msgs" + LOG.debug(reply) + return reply + + r = re.search(r"^\?-[fF]!", message) + if r is not None: + reply = "Deleting ALL Delayed msgs." + LOG.debug(reply) + tracker.flush() + return reply + + return reply diff --git a/aprsd/plugins/time.py b/aprsd/plugins/time.py new file mode 100644 index 0000000..3c0c3b9 --- /dev/null +++ b/aprsd/plugins/time.py @@ -0,0 +1,28 @@ +import logging +import time + +from aprsd import fuzzyclock, plugin + +LOG = logging.getLogger("APRSD") + + +class TimePlugin(plugin.APRSDPluginBase): + """Time command.""" + + version = "1.0" + command_regex = "^[tT]" + command_name = "time" + + def command(self, fromcall, message, ack): + LOG.info("TIME COMMAND") + stm = time.localtime() + h = stm.tm_hour + m = stm.tm_min + cur_time = fuzzyclock.fuzzy(h, m, 1) + reply = "{} ({}:{} PDT) ({})".format( + cur_time, + str(h), + str(m).rjust(2, "0"), + message.rstrip(), + ) + return reply diff --git a/aprsd/plugins/version.py b/aprsd/plugins/version.py new file mode 100644 index 0000000..d037ac7 --- /dev/null +++ b/aprsd/plugins/version.py @@ -0,0 +1,22 @@ +import logging + +import aprsd +from aprsd import plugin + +LOG = logging.getLogger("APRSD") + + +class VersionPlugin(plugin.APRSDPluginBase): + """Version of APRSD Plugin.""" + + version = "1.0" + command_regex = "^[vV]" + command_name = "version" + + # message_number:time combos so we don't resend the same email in + # five mins {int:int} + email_sent_dict = {} + + def command(self, fromcall, message, ack): + LOG.info("Version COMMAND") + return "APRSD version '{}'".format(aprsd.__version__) diff --git a/aprsd/plugins/weather.py b/aprsd/plugins/weather.py new file mode 100644 index 0000000..6071c1e --- /dev/null +++ b/aprsd/plugins/weather.py @@ -0,0 +1,53 @@ +import json +import logging + +from aprsd import plugin +import requests + +LOG = logging.getLogger("APRSD") + + +class WeatherPlugin(plugin.APRSDPluginBase): + """Weather Command""" + + version = "1.0" + command_regex = "^[wW]" + command_name = "weather" + + def command(self, fromcall, message, ack): + LOG.info("Weather Plugin") + try: + url = ( + "http://api.aprs.fi/api/get?" + "&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json" + "&name=%s" % fromcall + ) + response = requests.get(url) + # aprs_data = json.loads(response.read()) + aprs_data = json.loads(response.text) + lat = aprs_data["entries"][0]["lat"] + lon = aprs_data["entries"][0]["lng"] + url2 = ( + "https://forecast.weather.gov/MapClick.php?lat=%s" + "&lon=%s&FcstType=json" % (lat, lon) + ) + response2 = requests.get(url2) + # wx_data = json.loads(response2.read()) + wx_data = json.loads(response2.text) + reply = ( + "%sF(%sF/%sF) %s. %s, %s." + % ( + wx_data["currentobservation"]["Temp"], + wx_data["data"]["temperature"][0], + wx_data["data"]["temperature"][1], + wx_data["data"]["weather"][0], + wx_data["time"]["startPeriodName"][1], + wx_data["data"]["weather"][1], + ) + ).rstrip() + LOG.debug("reply: '{}' ".format(reply)) + except Exception as e: + LOG.debug("Weather failed with: " + "%s" % str(e)) + reply = "Unable to find you (send beacon?)" + + return reply diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 8512f34..48e3c44 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -2,8 +2,11 @@ import unittest from unittest import mock import aprsd -from aprsd import plugin from aprsd.fuzzyclock import fuzzy +from aprsd.plugins import fortune as fortune_plugin +from aprsd.plugins import ping as ping_plugin +from aprsd.plugins import time as time_plugin +from aprsd.plugins import version as version_plugin class TestPlugin(unittest.TestCase): @@ -14,17 +17,17 @@ class TestPlugin(unittest.TestCase): @mock.patch("shutil.which") def test_fortune_fail(self, mock_which): - fortune_plugin = plugin.FortunePlugin(self.config) + fortune = fortune_plugin.FortunePlugin(self.config) mock_which.return_value = None message = "fortune" expected = "Fortune command not installed" - actual = fortune_plugin.run(self.fromcall, message, self.ack) + actual = fortune.run(self.fromcall, message, self.ack) self.assertEqual(expected, actual) @mock.patch("subprocess.Popen") @mock.patch("shutil.which") def test_fortune_success(self, mock_which, mock_popen): - fortune_plugin = plugin.FortunePlugin(self.config) + fortune = fortune_plugin.FortunePlugin(self.config) mock_which.return_value = "/usr/bin/games" mock_process = mock.MagicMock() @@ -33,7 +36,7 @@ class TestPlugin(unittest.TestCase): message = "fortune" expected = "Funny fortune" - actual = fortune_plugin.run(self.fromcall, message, self.ack) + actual = fortune.run(self.fromcall, message, self.ack) self.assertEqual(expected, actual) @mock.patch("time.localtime") @@ -43,13 +46,13 @@ class TestPlugin(unittest.TestCase): m = fake_time.tm_min = 12 fake_time.tm_sec = 55 mock_time.return_value = fake_time - time_plugin = plugin.TimePlugin(self.config) + time = time_plugin.TimePlugin(self.config) fromcall = "KFART" message = "location" ack = 1 - actual = time_plugin.run(fromcall, message, ack) + actual = time.run(fromcall, message, ack) self.assertEqual(None, actual) cur_time = fuzzy(h, m, 1) @@ -61,7 +64,7 @@ class TestPlugin(unittest.TestCase): str(m).rjust(2, "0"), message.rstrip(), ) - actual = time_plugin.run(fromcall, message, ack) + actual = time.run(fromcall, message, ack) self.assertEqual(expected, actual) @mock.patch("time.localtime") @@ -72,7 +75,7 @@ class TestPlugin(unittest.TestCase): s = fake_time.tm_sec = 55 mock_time.return_value = fake_time - ping = plugin.PingPlugin(self.config) + ping = ping_plugin.PingPlugin(self.config) fromcall = "KFART" message = "location" @@ -102,19 +105,19 @@ class TestPlugin(unittest.TestCase): def test_version(self): expected = "APRSD version '{}'".format(aprsd.__version__) - version_plugin = plugin.VersionPlugin(self.config) + version = version_plugin.VersionPlugin(self.config) fromcall = "KFART" message = "No" ack = 1 - actual = version_plugin.run(fromcall, message, ack) + actual = version.run(fromcall, message, ack) self.assertEqual(None, actual) message = "version" - actual = version_plugin.run(fromcall, message, ack) + actual = version.run(fromcall, message, ack) self.assertEqual(expected, actual) message = "Version" - actual = version_plugin.run(fromcall, message, ack) + actual = version.run(fromcall, message, ack) self.assertEqual(expected, actual)