Laurent Kislaire 4 years ago
commit 2140a168b3
  1. 3
      README.md
  2. 46
      pyfbx/__main__.py
  3. 69
      pyfbx/client.py
  4. 2
      setup.py
  5. 10
      tests/conn_test.py

@ -156,7 +156,7 @@ pyfbx -c 'System.Get_the_current_system_info()' -c 'Connection.Get_the_current_C
Telegraf http listener v2 input plugin with json format can be used to plot data in realtime.
### Telegraf plots
### Telegraf and Graphana plots
You can use a telegraph configuration in /etc/telegraf/telegraf.d/freebox:
@ -174,7 +174,6 @@ You can use a telegraph configuration in /etc/telegraf/telegraf.d/freebox:
![Grafana](https://framagit.org/uploads/-/system/temp/7b198d06e9c27996b60708d470af96ae/debit.png "Grafana Dashboard")
## Developpment
### Testing

@ -9,6 +9,7 @@ import argparse
import pprint
import requests
from pyfbx import Fbx
from pyfbx.client import (ResponseError, RequestError)
log_level = (logging.WARNING, logging.INFO, logging.DEBUG)
@ -27,7 +28,7 @@ def console(log, level):
def output(fbx, command, json_output, send_url):
try:
res = eval("fbx.{}".format(command))
except Exception as exc:
except RequestError as exc:
if "403 Client Error" in str(exc):
fbx.log.info("Got 403, refreshing token")
fbx.mksession()
@ -38,7 +39,7 @@ def output(fbx, command, json_output, send_url):
try:
r = requests.post(send_url, json=res)
except BaseException as exc:
print("While sending to {}, got exception {}".format(send_url, exc))
log.error("While sending to {}, got exception {}".format(send_url, exc))
elif json_output:
print(json.dumps(res))
else:
@ -75,24 +76,31 @@ def main():
args.command = ["System.Get_the_current_system_info()"]
myfb = Fbx(nomdns=args.http, url=args.url)
if token:
if token.startswith('f:'):
with open(args.token[2:]) as tok_file:
token, app_id = tok_file.read().splitlines()
else:
log.warning("Registering app {}, id {}, Press button".format(__name__, app_id))
token = myfb.register(app_id=app_id, app_name=__name__,
device=os.uname().nodename)
log.warning("Save your application token: {}".format(token))
myfb.mksession(app_id=app_id, token=token)
try:
if token:
if token.startswith('f:'):
with open(args.token[2:]) as tok_file:
token, app_id = tok_file.read().splitlines()
else:
log.warning("Registering app {}, id {}, Press button".format(__name__, app_id))
token = myfb.register(app_id=app_id, app_name=__name__,
device=os.uname().nodename)
log.warning("Save your application token: {}".format(token))
myfb.mksession(app_id=app_id, token=token)
while True:
log.debug("\n")
for command in args.command:
output(myfb, command, args.json, args.send)
if not args.delay:
break
time.sleep(args.delay)
while True:
log.debug("-" * 8)
for command in args.command:
output(myfb, command, args.json, args.send)
if not args.delay:
break
time.sleep(args.delay)
except RequestError as exc:
log.error("Got Http Error {}".format(exc))
return 2
except ResponseError as exc:
log.error("Got Response Error {}".format(exc))
return 1
return 0

@ -1,4 +1,3 @@
#!/usr/bin/python3
"""
Fbx Client
"""
@ -13,10 +12,7 @@ from . import api
from . import utils
from . import mdns
__all__ = [
"FbxTransport",
"Fbx",
"FbxClass", ]
__all__ = ["Transport", "Fbx", "FbxClass"]
class FbxClass():
@ -28,7 +24,7 @@ class FbxClass():
self._trn = transport
class FbxTransport():
class Transport():
"""
Transport abstraction and context handling for all methods
"""
@ -53,23 +49,27 @@ class FbxTransport():
else:
self._url = url
def set_session_header(self, session_token):
def set_header(self, session_token):
self._session.headers.update({'X-Fbx-App-Auth': session_token})
def api_exec(self, http_method, endpoint, post_data=None, **kwargs):
req_response = self._session.request(
http_method,
self._url + "/" + endpoint.format(**kwargs), json=post_data)
self.log.debug(">> Sent %s %s/%s Post: %s", http_method, self._url,
endpoint.format(**kwargs), post_data)
req_response.raise_for_status()
try:
self.log.debug(">> Sent %s %s/%s Post: %s", http_method, self._url,
endpoint.format(**kwargs), post_data)
req_response = self._session.request(
http_method,
self._url + "/" + endpoint.format(**kwargs), json=post_data)
req_response.raise_for_status()
except requests.exceptions.RequestException as exc:
raise RequestError(exc)
response = req_response.json()
if response['success']:
if 'result' in response:
self.log.debug("<< Got {}".format(response['result']))
return response['result']
else:
raise FbxErrorResponse(response['error_code'], response['msg'])
raise ResponseError(response['error_code'], response['msg'])
def local_base(self, url=api._DISC_HTTP_URL):
response = self._session.get("{}/api_version".format(url)).json()
@ -85,9 +85,9 @@ class Fbx():
def __init__(self, url=None, nomdns=False, session=None):
self.log = logging.getLogger("pyfbx.fbx")
self._trn = FbxTransport(url, nomdns=nomdns, session=session)
self.app_id = None
self.token = None
self._trn = Transport(url, nomdns=nomdns, session=session)
self._app_id = None
self._token = None
# Create on the fly attributes to classes
_globals = globals()
@ -100,10 +100,10 @@ class Fbx():
"""
Register app
"""
self.app_id = app_id
data = {"app_id": self.app_id, "app_name": app_name, "device_name": device}
self._app_id = app_id
data = {"app_id": self._app_id, "app_name": app_name, "device_name": device}
res = self._trn.api_exec("POST", "login/authorize/", data)
trackid, self.token = res["track_id"], res["app_token"]
trackid, self._token = res["track_id"], res["app_token"]
s = "pending"
self.log.info("Press Ok on the freebox to register application")
while s == "pending":
@ -111,34 +111,41 @@ class Fbx():
if s == "pending":
time.sleep(1)
self.log.debug("Registration returned: {}".format(s))
return s == "granted" and self.token
return s == "granted" and self._token
def mksession(self, app_id=None, token=None):
self.log.debug("Making session with token={}[{}], app_id={}[{}]".format(
token, self.token, self.app_id, app_id))
token, self._token, app_id, self._app_id))
if token: # Don't overwrite previous token (used for refresh)
self.token = token
self._token = token
if app_id:
self.app_id = app_id
elif not self.app_id:
self._app_id = app_id
elif not self._app_id:
raise Exception("Missing app_id")
login = self._trn.api_exec("GET", "login/")
if not login['logged_in']:
data = {
"app_id": self.app_id,
"password": hmac.new(bytes(self.token, "ascii"),
"app_id": self._app_id,
"password": hmac.new(bytes(self._token, "ascii"),
bytes(login['challenge'], "ascii"),
hashlib.sha1).hexdigest()
}
resp = self._trn.api_exec("POST", "login/session/", data)
session_token = resp["session_token"]
self.app_id = app_id
self._trn.set_session_header(session_token)
self.log.info("Authenticated. Storing token={}, app_id={}".format(self.token, self.app_id))
self._trn.set_header(session_token)
self.log.info("Authenticated. Storing token={}, app_id={}".format(self._token, self._app_id))
return resp["permissions"]
class FbxErrorResponse(Exception):
class RequestError(Exception):
def __init__(self, msg):
self.msg = msg
def __str__(self):
return '{}'.format(self.msg)
class ResponseError(Exception):
def __init__(self, error_code, msg):
self.error_code = error_code
self.msg = msg

@ -3,7 +3,7 @@ from setuptools import setup, find_packages
setup(
name="pyfbx",
version="0.0.7",
version="0.0.8rc1",
description="Freebox thin client",
long_description=(pathlib.Path(__file__).parent / "README.md").read_text(),
long_description_content_type='text/markdown',

@ -8,7 +8,7 @@ import json
import pytest
from mock import (patch, PropertyMock, MagicMock)
from pyfbx import Fbx
from pyfbx.client import (FbxErrorResponse, FbxTransport)
from pyfbx.client import (ResponseError, Transport)
from pyfbx.mdns import FbxMDNS
@ -16,7 +16,7 @@ def test_fbx_register():
f = Fbx("http://192.168.1.254/api/v6")
with pytest.raises(Exception):
f.mksession()
with patch('pyfbx.client.FbxTransport.api_exec',
with patch('pyfbx.client.Transport.api_exec',
side_effect=iter([
{"track_id": "id", "app_token": "tok"},
{"status": "pending"},
@ -37,10 +37,10 @@ def test_fbx_session_local():
"first_name": "Sandy",
"last_name": "Kilo"})
f.Contacts.Delete_a_contact(id=r['id'])
with pytest.raises(FbxErrorResponse):
with pytest.raises(ResponseError):
try:
f.Contacts.Access_a_given_contact_entry(1)
except FbxErrorResponse as e:
except ResponseError as e:
print(e)
raise
with patch('requests.Response', PropertyMock) as mock_resp:
@ -66,7 +66,7 @@ def test_mdns():
def test_nonet():
with patch('pyfbx.client.FbxTransport._session', new_callable=PropertyMock, return_value=MagicMock(), create=True) as mock_session:
with patch('pyfbx.client.Transport._session', new_callable=PropertyMock, return_value=MagicMock(), create=True) as mock_session:
f = Fbx("http://192.168.12.34/api/v9")
f.System.Reboot_the_system()
f = Fbx()

Loading…
Cancel
Save