diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cfadd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1,39 @@ +*.py[co] + +#small test code that contains my keys +t.py + +# Packages +*.egg +*.egg-info +dist +build +eggs +parts +bin +include +lib +local +var +sdist +develop-eggs +.installed.cfg + +# Installer logs +pip-log.txt + +# Unit test / coverage reports +.coverage +.tox + +# PyCharm data +.idea + +#Translations +*.mo + +#Mr Developer +.mr.developer.cfg + +#Environment +env diff --git a/.travis.yml b/.travis.yml new file mode 100644 index 0000000..4348b43 --- /dev/null +++ b/.travis.yml @@ -0,0 +1,12 @@ +language: python +sudo: false +python: + - "2.7" + - "3.5" + +install: + - travis_retry pip install . + - pip install -r requirements.txt + +script: + - make test \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..0d068f3 --- /dev/null +++ b/Makefile @@ -0,0 +1,33 @@ +help: + @echo " env create a development environment using virtualenv" + @echo " deps install dependencies" + @echo " clean remove unwanted stuff" + @echo " test run tests" + +env: + sudo easy_install pip && \ + pip install virtualenv && \ + virtualenv env && \ + . env/bin/activate && \ + make deps + +deps: + pip install -r requirements.txt --use-mirrors + +clean: + rm -fr build + rm -fr dist + find . -name '*.pyc' -exec rm -f {} \; + find . -name '*.pyo' -exec rm -f {} \; + find . -name '*~' -exec rm -f {} \; + +test: + nosetests + +build: clean + python setup.py sdist + python setup.py bdist_wheel + +upload: clean + python setup.py sdist upload + python setup.py bdist_wheel upload diff --git a/README.md b/README.md new file mode 100644 index 0000000..50386be --- /dev/null +++ b/README.md @@ -0,0 +1,45 @@ +# Python Status.io + +A Python wrapper around the Status.io API. + +[![Downloads](https://img.shields.io/pypi/v/python-statusio.svg)](https://pypi.python.org/pypi/python-statusio/) +[![Travis CI](https://travis-ci.org/statusio/python-statusio.svg)](https://travis-ci.org/statusio/python-statusio) + +## Introduction + +This library provides a pure Python interface for the [Status.io API](http://docs.statusio.apiary.io/). It works with Python versions from 2.6+. + +[Status.io](http://status.io) provides hosted system status pages. + +## Installing + +You can install python-statusio using:: + + $ pip install python-statusio + +## Documentation + +View the last release API documentation at: http://developers.status.io/ + +## Using + +The library provides a Python wrapper around the Status.io API. + +### API + +The API is exposed via the `statusio.Api` class. + +To create an instance of the `statusio.Api` with yout credentials: + + >>> import statusio + >>> api = statusio.Api(api_id='api_id', + api_key='api_key') + +To your status page summary: + + >>> summary = api.StatusSummary('status_page_id') + >>> print(summary) + +There are many more API methods, to read the full API documentation:: + + $ pydoc statusio.Api \ No newline at end of file diff --git a/README.rst b/README.rst new file mode 100644 index 0000000..8f91f9c --- /dev/null +++ b/README.rst @@ -0,0 +1,58 @@ +Python Status.io +================ + +A Python wrapper around the Status.io API. + +|Downloads| |Travis CI| + +Introduction +------------ + +This library provides a pure Python interface for the `Status.io +API `__. It works with Python versions +from 2.6+. + +`Status.io `__ provides hosted system status pages. + +Installing +---------- + +You can install python-statusio using:: + + $ pip install python-statusio + +Documentation +------------- + +View the last release API documentation at: +http://docs.statusio.apiary.io/ + +Using +----- + +The library provides a Python wrapper around the Status.io API. + +API +~~~ + +The API is exposed via the ``statusio.Api`` class. + +To create an instance of the ``statusio.Api`` with yout credentials: + + >>> import statusio + >>> api = statusio.Api(api_id='api_id', + api_key='api_key') + +To your status page summary: + + >>> summary = api.StatusSummary('status_page_id') + >>> print(summary) + +There are many more API methods, to read the full API documentation:: + + $ pydoc statusio.Api + +.. |Downloads| image:: https://img.shields.io/pypi/v/python-statusio.svg + :target: https://pypi.python.org/pypi/python-statusio/ +.. |Travis CI| image:: https://travis-ci.org/statusio/python-statusio.svg + :target: https://travis-ci.org/statusio/python-statusio diff --git a/python-statusio.spec b/python-statusio.spec new file mode 100644 index 0000000..484a4a7 --- /dev/null +++ b/python-statusio.spec @@ -0,0 +1,48 @@ +%{!?python_sitelib: %define python_sitelib %(%{__python} -c "from distutils.sysconfig import get_python_lib; print get_python_lib()")} + +Name: python-statusio +Version: 0.1 +Release: %{?dist} +Summary: Python Interface for Status.io API + +Group: Development/Libraries +License: Apache License 2.0 +URL: http://github.com/statusio/python-statusio +BuildRoot: %{_tmppath}/%{name}-%{version}-%{release}-root-%(%{__id_u} -n) + +BuildArch: noarch +Requires: python >= 2.6, +BuildRequires: python-setuptools + + +%description +This library provides a pure python interface for the Status.io API. + + +%prep +%setup -q + + +%build +%{__python} setup.py build + + +%install +rm -rf $RPM_BUILD_ROOT +chmod a-x README +%{__python} setup.py install --skip-build --root $RPM_BUILD_ROOT + + +%clean +rm -rf $RPM_BUILD_ROOT + + +%files +%defattr(-,root,root,-) +%doc README.rst +# For noarch packages: sitelib +%{python_sitelib}/* + + +%changelog +- Initial package. \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9e9b9c7 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +future +requests \ No newline at end of file diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..957d9cd --- /dev/null +++ b/setup.cfg @@ -0,0 +1,7 @@ +[check-manifest] +ignore = + .travis.yml + violations.flake8.txt + +[flake8] +ignore = E111,E124,E126,E201,E202,E221,E241,E302,E501 \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..1fd4766 --- /dev/null +++ b/setup.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python + +'''The setup and build script for the python-statusio library.''' + +import os + +from setuptools import setup, find_packages + +def read(*paths): + """Build a file path from *paths* and return the contents.""" + with open(os.path.join(*paths), 'r') as f: + return f.read() + + +setup( + name='python-statusio', + version='0.2', + author='Status.io', + author_email='support@status.io', + license='Apache License 2.0', + url='https://github.com/statusio/python-statusio', + keywords='status.io api statusio', + description='A Python wrapper around the Status.io API', + long_description=(read('README.rst')), + packages=find_packages(exclude=['tests*']), + install_requires=['future', 'requests'], + classifiers=[ + 'Development Status :: 5 - Production/Stable', + 'Intended Audience :: Developers', + 'License :: OSI Approved :: Apache Software License', + 'Operating System :: OS Independent', + 'Topic :: Software Development :: Libraries :: Python Modules', + 'Topic :: Internet', + 'Programming Language :: Python', + 'Programming Language :: Python :: 2', + 'Programming Language :: Python :: 2.7', + ], +) diff --git a/statusio/__init__.py b/statusio/__init__.py new file mode 100644 index 0000000..dc2b4da --- /dev/null +++ b/statusio/__init__.py @@ -0,0 +1,11 @@ +#!/usr/bin/env python + +"""A library that provides a Python interface to the Status.io API""" +from __future__ import absolute_import + +__author__ = 'support@status.io' +__version__ = '0.2' + +import json # noqa + +from .api import Api # noqa diff --git a/statusio/api.py b/statusio/api.py new file mode 100644 index 0000000..b77fbdc --- /dev/null +++ b/statusio/api.py @@ -0,0 +1,1007 @@ +#!/usr/bin/env python + +"""A library that provides a Python interface to the Status.io API""" +from __future__ import division +from __future__ import print_function + +import sys +import gzip +import time +import types +import base64 +import re +import datetime +from calendar import timegm +import requests +import io + +from past.utils import old_div + +try: + # python 3 + from urllib.parse import urlparse, urlunparse, urlencode + from urllib.request import urlopen + from urllib.request import __version__ as urllib_version +except ImportError: + from urlparse import urlparse, urlunparse + from urllib2 import urlopen + from urllib import urlencode + from urllib import __version__ as urllib_version + +from statusio import (__version__, json) + + +class Api(object): + """A python interface into the Status.io API + + Example usage: + + Simples example + + >>> import statusio + >>> api = statusio.Api(API_ID, API_KEY) + >>> result = api.StatusSummary(STATUSPAGE_ID) + >>> print(result) + + There are many other methods, including: + + >>> api.ComponentList(statuspage_id) + >>> api.ComponentStatusUpdate(statuspage_id, components, containers, details, current_status) + >>> api.IncidentList(statuspage_id) + >>> api.IncidentMessage(statuspage_id, message_id) + >>> api.IncidentCreate(statuspage_id, components, containers, incident_name, incident_details, current_status, current_state, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0, all_infrastructure_affected=0) + >>> api.IncidentUpdate(statuspage_id, incident_id, incident_details, current_status, current_state, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0) + >>> api.IncidentResolve(statuspage_id, incident_id, incident_details, current_status, current_state, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0) + >>> api.IncidentDelete(statuspage_id, incident_id) + >>> api.MaintenanceList(statuspage_id) + >>> api.MaintenanceMessage(statuspage_id, message_id) + >>> api.MaintenanceSchedule(statuspage_id, components, containers, maintenance_name, maintenance_details, date_planned_start, time_planned_start, date_planned_end, time_planned_end, automation=0, all_infrastructure_affected=0, maintenance_notify_now=0, maintenance_notify_1_hr=0, maintenance_notify_24_hr=0, maintenance_notify_72_hr=0) + >>> api.MaintenanceStart(statuspage_id, maintenance_id, maintenance_details, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0) + >>> api.MaintenanceUpdate(statuspage_id, maintenance_id, maintenance_details, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0) + >>> api.MaintenanceFinish(statuspage_id, maintenance_id, maintenance_details, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0) + >>> api.MaintenanceDelete(statuspage_id, maintenance_id) + >>> api.MetricUpdate(statuspage_id, metric_id, day_avg, day_start, day_dates, day_values, week_avg, week_start, week_dates, week_values, month_avg, month_start, month_dates, month_values) + >>> api.StatusSummary(statuspage_id) + >>> api.SubscriberList(statuspage_id) + >>> api.SubscriberAdd(statuspage_id, method, address, silent=1, granular='') + >>> api.SubscriberUpdate(statuspage_id, subscriber_id, address, granular='') + >>> api.SubscriberRemove(statuspage_id, subscriber_id) + """ + + def __init__(self, + api_id, + api_key, + version=2, + base_url='https://api.status.io' + ): + """Instantiate a new statusio.Api object. + + Args: + api_id: + Your Status.io API ID. + api_key: + Your Status.io API KEY. + version: + API version number. [Optional] + base_url: + API base URL. [Optional] + """ + self._api_id = api_id + self._api_key = api_key + self.base_url = '%s/v%d' % (base_url, version) + + def ComponentList(self, statuspage_id): + """List all components. + + Args: + statuspage_id: + Status page ID + + Returns: + A JSON object. + """ + url = '%s/component/list/%s' % (self.base_url, statuspage_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def ComponentStatusUpdate(self, + statuspage_id, + components, + containers, + details, + current_status): + """Update the status of a component on the fly without creating an incident or maintenance. + + Args: + statuspage_id: + Status page ID + components: + ID of each affected component + containers: + ID of each affected container + details: + A brief message describing this update + current_status: + Any numeric status code. + + Returns: + A JSON object. + """ + url = '%s/component/status/update' % self.base_url + resp = self._RequestUrl(url, 'POST', data={ + 'statuspage_id': statuspage_id, + 'components': components, + 'containers': containers, + 'details': details, + 'current_status': current_status + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def IncidentList(self, statuspage_id): + """List all active and resolved incidents. + + Args: + statuspage_id: + Status page ID + + Returns: + A JSON object. + """ + url = '%s/incident/list/%s' % (self.base_url, statuspage_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def IncidentMessage(self, + statuspage_id, + message_id): + """Display incident message. + + Args: + statuspage_id: + Status page ID + message_id: + Message ID + + Returns: + A JSON object. + """ + url = '%s/incident/message/%s/%s' % (self.base_url, + statuspage_id, message_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def IncidentCreate(self, + statuspage_id, + components, + containers, + incident_name, + incident_details, + current_status, + current_state, + notify_email=0, + notify_sms=0, + notify_webhook=0, + social=0, + irc=0, + hipchat=0, + slack=0, + all_infrastructure_affected=0): + """Create a new incident. + + Args: + statuspage_id: + Status page ID + components: + ID of each affected component + containers: + ID of each affected container + incident_name: + A descriptive title for the incident + incident_details: + Message describing this incident + current_status: + The status of the components and containers affected by this incident + current_state: + The state of this incident + notify_email: + Notify email subscribers (1 = Send notification) + notify_sms: + Notify SMS subscribers (1 = Send notification) + notify_webhook: + Notify webhook subscribers (1 = Send notification) + social: + Automatically Tweet this update. (1 = Send Tweet) + irc: + Notify IRC channel (1 = Send notification) + hipchat: + Notify HipChat room (1 = Send notification) + slack: + Notify Slack channel (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/incident/create' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'components': components, + 'containers': containers, + 'incident_name': incident_name, + 'incident_details': incident_details, + 'current_status': current_status, + 'current_state': current_state, + 'notify_email': notify_email, + 'notify_sms': notify_sms, + 'notify_webhook': notify_webhook, + 'social': social, + 'irc': irc, + 'hipchat': hipchat, + 'slack': slack, + 'all_infrastructure_affected': all_infrastructure_affected + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def IncidentUpdate(self, + statuspage_id, + incident_id, + incident_details, + current_status, + current_state, + notify_email=0, + notify_sms=0, + notify_webhook=0, + social=0, + irc=0, + hipchat=0, + slack=0): + """Update an existing incident + + Args: + statuspage_id: + Status page ID + incident_id: + Incident ID + incident_details: + Message describing this incident + current_status: + The status of the components and containers affected by this incident + current_state: + The state of this incident + notify_email: + Notify email subscribers (1 = Send notification) + notify_sms: + Notify SMS subscribers (1 = Send notification) + notify_webhook: + Notify webhook subscribers (1 = Send notification) + social: + Automatically Tweet this update. (1 = Send Tweet) + irc: + Notify IRC channel (1 = Send notification) + hipchat: + Notify HipChat room (1 = Send notification) + slack: + Notify Slack channel (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/incident/update' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'incident_id': incident_id, + 'incident_details': incident_details, + 'current_status': current_status, + 'current_state': current_state, + 'notify_email': notify_email, + 'notify_sms': notify_sms, + 'notify_webhook': notify_webhook, + 'social': social, + 'irc': irc, + 'hipchat': hipchat, + 'slack': slack + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def IncidentResolve(self, + statuspage_id, + incident_id, + incident_details, + current_status, + current_state, + notify_email=0, + notify_sms=0, + notify_webhook=0, + social=0, + irc=0, + hipchat=0, + slack=0): + """Resolve an existing incident. The incident will be shown in the history instead of on the main page. + + Args: + statuspage_id: + Status page ID + incident_id: + Incident ID + incident_details: + Message describing this incident + current_status: + The status of the components and containers affected by this incident + current_state: + The state of this incident + notify_email: + Notify email subscribers (1 = Send notification) + notify_sms: + Notify SMS subscribers (1 = Send notification) + notify_webhook: + Notify webhook subscribers (1 = Send notification) + social: + Automatically Tweet this update. (1 = Send Tweet) + irc: + Notify IRC channel (1 = Send notification) + hipchat: + Notify HipChat room (1 = Send notification) + slack: + Notify Slack channel (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/incident/resolve' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'incident_id': incident_id, + 'incident_details': incident_details, + 'current_status': current_status, + 'current_state': current_state, + 'notify_email': notify_email, + 'notify_sms': notify_sms, + 'notify_webhook': notify_webhook, + 'social': social, + 'irc': irc, + 'hipchat': hipchat, + 'slack': slack + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def IncidentDelete(self, + statuspage_id, + incident_id): + """Delete an existing incident. The incident will be deleted forever and cannot be recovered. + + Args: + statuspage_id: + Status page ID + incident_id: + Incident ID + + Returns: + A JSON object. + """ + url = '%s/incident/delete' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'incident_id': incident_id + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceList(self, statuspage_id): + """List all active, resolved and upcoming maintenances + + Args: + statuspage_id: + Status page ID + + Returns: + A JSON object. + """ + url = '%s/maintenance/list/%s' % (self.base_url, statuspage_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceMessage(self, + statuspage_id, + message_id): + """Display maintenance message + + Args: + statuspage_id: + Status page ID + incident_id: + Message ID + + Returns: + A JSON object. + """ + url = '%s/maintenance/message/%s/%s' % ( + self.base_url, statuspage_id, message_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceSchedule(self, + statuspage_id, + components, + containers, + maintenance_name, + maintenance_details, + date_planned_start, + time_planned_start, + date_planned_end, + time_planned_end, + automation=0, + all_infrastructure_affected=0, + maintenance_notify_now=0, + maintenance_notify_1_hr=0, + maintenance_notify_24_hr=0, + maintenance_notify_72_hr=0): + """Schedule a new maintenance + + Args: + statuspage_id: + Status page ID + components: + ID of each affected component + containers: + ID of each affected container + maintenance_name: + A descriptive title for this maintenance + maintenance_details: + Message describing this maintenance + date_planned_start: + Date maintenance is expected to start + time_planned_start: + Time maintenance is expected to start + date_planned_end: + Date maintenance is expected to end + time_planned_end: + Time maintenance is expected to end + automation: + Automatically start and end the maintenance (default = 0) + all_infrastructure_affected: + Affect all components and containers (default = 0) + maintenance_notify_now: + Notify subscribers now (1 = Send notification) + maintenance_notify_1_hr: + Notify subscribers 1 hour before scheduled maintenance start time (1 = Send notification) + maintenance_notify_24_hr: + Notify subscribers 24 hours before scheduled maintenance start time (1 = Send notification) + maintenance_notify_72_hr: + Notify subscribers 72 hours before scheduled maintenance start time (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/maintenance/schedule' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'components': components, + 'containers': containers, + 'maintenance_name': maintenance_name, + 'maintenance_details': maintenance_details, + 'date_planned_start': date_planned_start, + 'time_planned_start': time_planned_start, + 'date_planned_end': date_planned_end, + 'time_planned_end': time_planned_end, + 'automation': automation, + 'all_infrastructure_affected': all_infrastructure_affected, + 'maintenance_notify_now': maintenance_notify_now, + 'maintenance_notify_1_hr': maintenance_notify_1_hr, + 'maintenance_notify_24_hr': maintenance_notify_24_hr, + 'maintenance_notify_72_hr': maintenance_notify_72_hr + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceStart(self, + statuspage_id, + maintenance_id, + maintenance_details, + notify_email=0, + notify_sms=0, + notify_webhook=0, + social=0, + irc=0, + hipchat=0, + slack=0): + """Begin a scheduled maintenance now + + Args: + statuspage_id: + Status page ID + maintenance_id: + Maintenance ID + maintenance_details: + Message describing this maintenance update + notify_email: + Notify email subscribers (1 = Send notification) + notify_sms: + Notify SMS subscribers (1 = Send notification) + notify_webhook: + Notify webhook subscribers (1 = Send notification) + social: + Automatically Tweet this update. (1 = Send Tweet) + irc: + Notify IRC channel (1 = Send notification) + hipchat: + Notify HipChat room (1 = Send notification) + slack: + Notify Slack channel (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/maintenance/start' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'maintenance_id': maintenance_id, + 'maintenance_details': maintenance_details, + 'notify_email': notify_email, + 'notify_sms': notify_sms, + 'notify_webhook': notify_webhook, + 'social': social, + 'irc': irc, + 'hipchat': hipchat, + 'slack': slack + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceUpdate(self, + statuspage_id, + maintenance_id, + maintenance_details, + notify_email=0, + notify_sms=0, + notify_webhook=0, + social=0, + irc=0, + hipchat=0, + slack=0): + """Update an active maintenance + + Args: + statuspage_id: + Status page ID + maintenance_id: + Maintenance ID + maintenance_details: + Message describing this maintenance update + notify_email: + Notify email subscribers (1 = Send notification) + notify_sms: + Notify SMS subscribers (1 = Send notification) + notify_webhook: + Notify webhook subscribers (1 = Send notification) + social: + Automatically Tweet this update. (1 = Send Tweet) + irc: + Notify IRC channel (1 = Send notification) + hipchat: + Notify HipChat room (1 = Send notification) + slack: + Notify Slack channel (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/maintenance/update' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'maintenance_id': maintenance_id, + 'maintenance_details': maintenance_details, + 'notify_email': notify_email, + 'notify_sms': notify_sms, + 'notify_webhook': notify_webhook, + 'social': social, + 'irc': irc, + 'hipchat': hipchat, + 'slack': slack + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceFinish(self, + statuspage_id, + maintenance_id, + maintenance_details, + notify_email=0, + notify_sms=0, + notify_webhook=0, + social=0, + irc=0, + hipchat=0, + slack=0): + """Close an active maintenance. The maintenance will be moved to the history. + + Args: + statuspage_id: + Status page ID + maintenance_id: + Maintenance ID + maintenance_details: + Message describing this maintenance update + notify_email: + Notify email subscribers (1 = Send notification) + notify_sms: + Notify SMS subscribers (1 = Send notification) + notify_webhook: + Notify webhook subscribers (1 = Send notification) + social: + Automatically Tweet this update. (1 = Send Tweet) + irc: + Notify IRC channel (1 = Send notification) + hipchat: + Notify HipChat room (1 = Send notification) + slack: + Notify Slack channel (1 = Send notification) + + Returns: + A JSON object. + """ + url = '%s/maintenance/finish' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'maintenance_id': maintenance_id, + 'maintenance_details': maintenance_details, + 'notify_email': notify_email, + 'notify_sms': notify_sms, + 'notify_webhook': notify_webhook, + 'social': social, + 'irc': irc, + 'hipchat': hipchat, + 'slack': slack + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def MaintenanceDelete(self, + statuspage_id, + maintenance_id): + """Delete an existing maintenance. The maintenance will be deleted forever and cannot be recovered. + + Args: + statuspage_id: + Status page ID + maintenance_id: + Maintenance ID + + Returns: + A JSON object. + """ + url = '%s/maintenance/delete' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'maintenance_id': maintenance_id + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def MetricUpdate(self, + statuspage_id, + metric_id, + day_avg, + day_start, + day_dates, + day_values, + week_avg, + week_start, + week_dates, + week_values, + month_avg, + month_start, + month_dates, + month_values): + """Update custom metric data + + Args: + statuspage_id: + Status page ID + metric_id: + Metric ID + day_avg: + Average value for past 24 hours + day_start: + UNIX timestamp for start of metric timeframe + day_dates: + An array of timestamps for the past 24 hours (2014-03-28T05:43:00+00:00) + day_values: + An array of values matching the timestamps (Must be 24 values) + week_avg: + Average value for past 7 days + week_start: + UNIX timestamp for start of metric timeframe + week_dates: + An array of timestamps for the past 7 days (2014-03-28T05:43:00+00:00) + week_values: + An array of values matching the timestamps (Must be 7 values) + month_avg: + Average value for past 30 days + month_start: + UNIX timestamp for start of metric timeframe + month_dates: + An array of timestamps for the past 30 days (2014-03-28T05:43:00+00:00) + month_values: + An array of values matching the timestamps (Must be 30 values) + + Returns: + A JSON object. + """ + url = '%s/metric/update' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'metric_id': metric_id, + 'day_avg': day_avg, + 'day_start': day_start, + 'day_dates': day_dates, + 'day_values': day_values, + 'week_avg': week_avg, + 'week_start': week_start, + 'week_dates': week_dates, + 'week_values': week_values, + 'month_avg': month_avg, + 'month_start': month_start, + 'month_dates': month_dates, + 'month_values': month_values + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def StatusSummary(self, statuspage_id): + """Show the summary status for all components and containers + + Args: + statuspage_id: + Status page ID + + Returns: + A JSON object. + """ + url = '%s/status/summary/%s' % (self.base_url, statuspage_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def SubscriberList(self, statuspage_id): + """List all subscribers + + Args: + statuspage_id: + Status page ID + + Returns: + A JSON object. + """ + url = '%s/subscriber/list/%s' % (self.base_url, statuspage_id) + resp = self._RequestUrl(url, 'GET') + data = json.loads(resp.content.decode('utf-8')) + return data + + def SubscriberAdd(self, + statuspage_id, + method, + address, + silent=1, + granular=''): + """Add a new subscriber + + Args: + statuspage_id: + Status page ID + method: + Communication method of subscriber. Valid methods are `email`, `sms` or `webhook` + address: + Subscriber address (SMS number must include country code ie. +1) + silent: + Supress the welcome message (1 = Do not send notification) + granular: + List of component_container combos + + Returns: + A JSON object. + """ + url = '%s/subscriber/add' % self.base_url + resp = self._RequestUrl(url, 'POST', { + 'statuspage_id': statuspage_id, + 'method': method, + 'address': address, + 'silent': silent, + 'granular': granular, + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def SubscriberUpdate(self, + statuspage_id, + subscriber_id, + address, + granular=''): + """Update existing subscriber + + Args: + statuspage_id: + Status page ID + subscriber_id: + SubscriberAdd ID + address: + Subscriber address (SMS number must include country code ie. +1) + granular: + List of component_container combos + + Returns: + A JSON object. + """ + url = '%s/subscriber/update' % self.base_url + resp = self._RequestUrl(url, 'PATCH', { + 'statuspage_id': statuspage_id, + 'subscriber_id': subscriber_id, + 'address': address, + 'granular': granular, + }) + data = json.loads(resp.content.decode('utf-8')) + return data + + def SubscriberRemove(self, + statuspage_id, + subscriber_id): + """Delete subscriber + + Args: + statuspage_id: + Status page ID + subscriber_id: + Subscriber ID + + Returns: + A JSON object. + """ + url = '%s/subscriber/remove/%s/%s' % ( + self.base_url, statuspage_id, subscriber_id) + resp = self._RequestUrl(url, 'DELETE') + data = json.loads(resp.content.decode('utf-8')) + return data + + def _BuildUrl(self, url, path_elements=None, extra_params=None): + # Break url into constituent parts + (scheme, netloc, path, params, query, fragment) = urlparse(url) + + # Add any additional path elements to the path + if path_elements: + # Filter out the path elements that have a value of None + p = [i for i in path_elements if i] + if not path.endswith('/'): + path += '/' + path += '/'.join(p) + + # Add any additional query parameters to the query string + if extra_params and len(extra_params) > 0: + extra_query = self._EncodeParameters(extra_params) + # Add it to the existing query + if query: + query += '&' + extra_query + else: + query = extra_query + + # Return the rebuilt URL + return urlunparse((scheme, netloc, path, params, query, fragment)) + + def _Encode(self, s): + if self._input_encoding: + return str(s, self._input_encoding).encode('utf-8') + else: + return str(s).encode('utf-8') + + def _EncodeParameters(self, parameters): + """Return a string in key=value&key=value form. + + Values of None are not included in the output string. + + Args: + parameters: + A dict of (key, value) tuples, where value is encoded as + specified by self._encoding + + Returns: + A URL-encoded string in "key=value&key=value" form + """ + if parameters is None: + return None + else: + return urlencode(dict([(k, self._Encode(v)) for k, v in list( + parameters.items()) if v is not None])) + + def _EncodePostData(self, post_data): + """Return a string in key=value&key=value form. + + Values are assumed to be encoded in the format specified by self._encoding, + and are subsequently URL encoded. + + Args: + post_data: + A dict of (key, value) tuples, where value is encoded as + specified by self._encoding + + Returns: + A URL-encoded string in "key=value&key=value" form + """ + if post_data is None: + return None + else: + return urlencode(dict([(k, self._Encode(v)) + for k, v in list(post_data.items())])) + + def _RequestUrl(self, url, verb, data=None): + """Request a url. + + Args: + url: + The web location we want to retrieve. + verb: + Either POST or GET. + data: + A dict of (str, unicode) key/value pairs. + + Returns: + A JSON object. + """ + if verb == 'POST': + try: + return requests.post( + url, + data=json.dumps(data), + headers={ + 'x-api-id': self._api_id, + 'x-api-key': self._api_key, + 'content-type': 'application/json' + } + ) + except requests.RequestException as e: + raise Error(str(e)) + elif verb == 'GET': + url = self._BuildUrl(url, extra_params=data) + try: + return requests.get( + url, + headers={ + 'x-api-id': self._api_id, + 'x-api-key': self._api_key + } + ) + except requests.RequestException as e: + raise Error(str(e)) + elif verb == 'PATCH': + try: + return requests.patch( + url, + data=json.dumps(data), + headers={ + 'x-api-id': self._api_id, + 'x-api-key': self._api_key, + 'content-type': 'application/json' + } + ) + except requests.RequestException as e: + raise Error(str(e)) + elif verb == 'DELETE': + url = self._BuildUrl(url, extra_params=data) + try: + return requests.delete( + url, + headers={ + 'x-api-id': self._api_id, + 'x-api-key': self._api_key, + 'content-type': 'application/json' + } + ) + except requests.RequestException as e: + raise Error(str(e)) + + return 0 diff --git a/test.py b/test.py new file mode 100644 index 0000000..4ef1aac --- /dev/null +++ b/test.py @@ -0,0 +1,5 @@ +import unittest + +if __name__ == '__main__': + testsuite = unittest.TestLoader().discover('.') + unittest.TextTestRunner(verbosity=1).run(testsuite) \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..49439c6 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,340 @@ +# encoding: utf-8 + +import os +import time +import unittest +import statusio + +API_ID = os.environ.get('API_ID') +API_KEY = os.environ.get('API_KEY') +STATUSPAGE_ID = os.environ.get('STATUSPAGE_ID') +COMPONENTS = [os.environ.get('COMPONENT')] +CONTAINERS = [os.environ.get('CONTAINER')] +METRIC_ID = os.environ.get('METRIC_ID') + +ID1 = '' +ID2 = '' + + +class ApiTest(unittest.TestCase): + + def setUp(self): + self._api = statusio.Api(API_ID, API_KEY) + + # STATUS + + def testStatus1Summary(self): + # Test the statusio.StatusSummary method + print('Testing StatusSummary') + data = self._api.StatusSummary(STATUSPAGE_ID) + self.assertEqual(data['status']['error'], 'no') + + # SUBSCRIBER + + def testSubscriber1Add(self): + # Test the statusio.SubscriberAdd method + print('Testing SubscriberAdd') + global ID1, ID2 + data = self._api.SubscriberAdd( + STATUSPAGE_ID, "email", "phillip.j.fry@planet-express12.com") + self.assertEqual(data['status']['error'], 'no') + ID1 = data['subscriber_id'] + print(data['subscriber_id']) + print(ID1) + + def testSubscriber2List(self): + # Test the statusio.SubscriberList method + print('Testing SubscriberList') + global ID1, ID2 + data = self._api.SubscriberList(STATUSPAGE_ID) + self.assertEqual(data['status']['error'], 'no') + self.assertEqual(data['result']['email'][0]['_id'], ID1) + + def testSubscriber3Update(self): + # Test the statusio.SubscriberUpdate method + print('Testing SubscriberUpdate') + global ID1, ID2 + data = self._api.SubscriberUpdate( + STATUSPAGE_ID, ID1, "phillip.j.fry.1@planet-express.com") + self.assertEqual(data['status']['error'], 'no') + + def testSubscriber4Remove(self): + # Test the statusio.SubscriberRemove method + print('Testing SubscriberRemove') + global ID1, ID2 + data = self._api.SubscriberRemove(STATUSPAGE_ID, ID1) + self.assertEqual(data['status']['error'], 'no') + + # MAINTENANCE + + def testMaintenance1Schedule(self): + # Test the statusio.MaintenanceSchedule method + print('Testing MaintenanceSchedule') + global ID1, ID2 + data = self._api.MaintenanceSchedule( + STATUSPAGE_ID, + COMPONENTS, + CONTAINERS, + 'Autotest', + 'Autotest Description', + '2018/12/31', + '23:59', + '2019/01/01', + '23:59') + self.assertEqual(data['status']['error'], 'no') + ID1 = data['result'] + + def testMaintenance2List(self): + # Test the statusio.MaintenanceList method + print('Testing MaintenanceList') + global ID1, ID2 + data = self._api.MaintenanceList(STATUSPAGE_ID) + self.assertEqual(data['status']['error'], 'no') + self.assertEqual( + data['result']['upcoming_maintenances'][0]['_id'], ID1) + ID2 = data['result']['upcoming_maintenances'][0]['messages'][0]['_id'] + + def testMaintenance3Message(self): + # Test the statusio.MaintenanceMessage method + print('Testing MaintenanceMessage') + global ID1, ID2 + data = self._api.MaintenanceMessage(STATUSPAGE_ID, ID2) + self.assertEqual(data['status']['error'], 'no') + + def testMaintenance4Start(self): + # Test the statusio.MaintenanceStart method + print('Testing MaintenanceStart') + global ID1, ID2 + data = self._api.MaintenanceStart( + STATUSPAGE_ID, ID1, 'Autotest details') + self.assertEqual(data['status']['error'], 'no') + + def testMaintenance5Update(self): + # Test the statusio.MaintenanceUpdate method + print('Testing MaintenanceUpdate') + global ID1, ID2 + data = self._api.MaintenanceUpdate( + STATUSPAGE_ID, ID1, 'Autotest details update') + self.assertEqual(data['status']['error'], 'no') + + def testMaintenance7Finish(self): + # Test the statusio.MaintenanceFinish method + print('Testing MaintenanceFinish') + global ID1, ID2 + data = self._api.MaintenanceFinish( + STATUSPAGE_ID, ID1, 'Autotest details finish') + self.assertEqual(data['status']['error'], 'no') + + def testMaintenance8Delete(self): + # Test the statusio.MaintenanceDelete method + print('Testing MaintenanceDelete') + global ID1, ID2 + data = self._api.MaintenanceDelete(STATUSPAGE_ID, ID1) + self.assertEqual(data['status']['error'], 'no') + + # INCIDENT + + def testIncident1Create(self): + # Test the statusio.IncidentCreate method + print('Testing IncidentCreate') + global ID1, ID2 + data = self._api.IncidentCreate( + STATUSPAGE_ID, + COMPONENTS, + CONTAINERS, + 'Autotest', + 'Autotest details', + 300, + 100) + self.assertEqual(data['status']['error'], 'no') + ID1 = data['result'] + + def testIncident2List(self): + # Test the statusio.IncidentList method + print('Testing IncidentList') + global ID1, ID2 + data = self._api.IncidentList(STATUSPAGE_ID) + self.assertEqual(data['status']['error'], 'no') + self.assertEqual(data['result']['active_incidents'][0]['_id'], ID1) + ID2 = data['result']['active_incidents'][0]['messages'][0]['_id'] + + def testIncident3Message(self): + # Test the statusio.IncidentMessage method + print('Testing IncidentMessage') + global ID1, ID2 + data = self._api.IncidentMessage(STATUSPAGE_ID, ID2) + self.assertEqual(data['status']['error'], 'no') + + def testIncident5Update(self): + # Test the statusio.IncidentUpdate method + print('Testing IncidentUpdate') + global ID1, ID2 + data = self._api.IncidentUpdate( + STATUSPAGE_ID, ID1, 'Autotest details update', 300, 100) + self.assertEqual(data['status']['error'], 'no') + + def testIncident7Resolve(self): + # Test the statusio.IncidentResolve method + print('Testing IncidentResolve') + global ID1, ID2 + data = self._api.IncidentResolve( + STATUSPAGE_ID, ID1, 'Autotest details resolve', 300, 100) + self.assertEqual(data['status']['error'], 'no') + + def testIncident8Delete(self): + # Test the statusio.IncidentDelete method + print('Testing IncidentDelete') + global ID1, ID2 + data = self._api.IncidentDelete(STATUSPAGE_ID, ID1) + self.assertEqual(data['status']['error'], 'no') + + # METRIC + + def testMetric1Update(self): + # Test the statusio.MetricUpdate method + print('Testing MetricUpdate') + global ID1, ID2 + data = self._api.MetricUpdate(STATUSPAGE_ID, + METRIC_ID, + 20.69, + 1395981878000, + ["2014-03-28T05:43:00+00:00", + "2014-03-28T06:43:00+00:00", + "2014-03-28T07:43:00+00:00", + "2014-03-28T08:43:00+00:00", + "2014-03-28T09:43:00+00:00", + "2014-03-28T10:43:00+00:00", + "2014-03-28T11:43:00+00:00", + "2014-03-28T12:43:00+00:00", + "2014-03-28T13:43:00+00:00", + "2014-03-28T14:43:00+00:00", + "2014-03-28T15:43:00+00:00", + "2014-03-28T16:43:00+00:00", + "2014-03-28T17:43:00+00:00", + "2014-03-28T18:43:00+00:00", + "2014-03-28T19:43:00+00:00", + "2014-03-28T20:43:00+00:00", + "2014-03-28T21:43:00+00:00", + "2014-03-28T22:43:00+00:00", + "2014-03-28T23:43:00+00:00", + "2014-03-29T00:43:00+00:00", + "2014-03-29T01:43:00+00:00", + "2014-03-29T02:43:00+00:00", + "2014-03-29T03:43:00+00:00"], + [20.70, + 20.00, + 19.20, + 19.80, + 19.90, + 20.10, + 21.40, + 23.00, + 27.40, + 28.70, + 27.50, + 29.30, + 28.50, + 27.20, + 28.60, + 28.70, + 25.90, + 23.40, + 22.40, + 21.40, + 19.80, + 19.50, + 20.00], + 20.07, + 1395463478000, + ["2014-03-22T04:43:00+00:00", + "2014-03-23T04:43:00+00:00", + "2014-03-24T04:43:00+00:00", + "2014-03-25T04:43:00+00:00", + "2014-03-26T04:43:00+00:00", + "2014-03-27T04:43:00+00:00", + "2014-03-28T04:43:00+00:00"], + [23.10, + 22.10, + 22.20, + 22.30, + 22.10, + 18.70, + 17.00], + 10.63, + 1393476280000, + ["2014-02-28T04:43:00+00:00", + "2014-03-01T04:43:00+00:00", + "2014-03-02T04:43:00+00:00", + "2014-03-03T04:43:00+00:00", + "2014-03-04T04:43:00+00:00", + "2014-03-05T04:43:00+00:00", + "2014-03-06T04:43:00+00:00", + "2014-03-07T04:43:00+00:00", + "2014-03-08T04:43:00+00:00", + "2014-03-09T04:43:00+00:00", + "2014-03-10T04:43:00+00:00", + "2014-03-11T04:43:00+00:00", + "2014-03-12T04:43:00+00:00", + "2014-03-13T04:43:00+00:00", + "2014-03-14T04:43:00+00:00", + "2014-03-15T04:43:00+00:00", + "2014-03-16T04:43:00+00:00", + "2014-03-17T04:43:00+00:00", + "2014-03-18T04:43:00+00:00", + "2014-03-19T04:43:00+00:00", + "2014-03-20T04:43:00+00:00", + "2014-03-21T04:43:00+00:00", + "2014-03-22T04:43:00+00:00", + "2014-03-23T04:43:00+00:00", + "2014-03-24T04:43:00+00:00", + "2014-03-25T04:43:00+00:00", + "2014-03-26T04:43:00+00:00", + "2014-03-27T04:43:00+00:00", + "2014-03-28T04:43:00+00:00"], + [0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 0.00, + 18.50, + 18.60, + 18.40, + 16.60, + 16.80, + 17.90, + 19.90, + 21.30, + 22.80, + 20.00, + 17.30, + 19.10, + 21.50, + 22.40, + 22.50, + 22.00, + 21.80]) + self.assertEqual(data['status']['error'], 'no') + + # COMPONENT + + def testComponent1List(self): + # Test the statusio.ComponentList method + print('Testing ComponentList') + global ID1, ID2 + data = self._api.ComponentList(STATUSPAGE_ID) + self.assertEqual(data['status']['error'], 'no') + + def testComponent2StatusUpdate(self): + # Test the statusio.ComponentStatusUpdate method + print('Testing ComponentStatusUpdate') + global ID1, ID2 + data = self._api.ComponentStatusUpdate( + STATUSPAGE_ID, COMPONENTS, CONTAINERS, 'Test status', 300) + self.assertEqual(data['status']['error'], 'no')