Browse Source

Merge branch 'dev' into 'master'

0.0.5

See merge request sun/pyfbx!12
tags/v0.0.5
Laurent 1 year ago
parent
commit
bec6fb5683
9 changed files with 149 additions and 133 deletions
  1. +1
    -1
      .gitlab-ci.yml
  2. +16
    -8
      README.md
  3. +14
    -5
      pyfbx/__main__.py
  4. +102
    -102
      pyfbx/api.py
  5. +8
    -9
      pyfbx/client.py
  6. +2
    -2
      pyfbx/mdns.py
  7. +4
    -4
      pyfbx/utils.py
  8. +1
    -1
      setup.py
  9. +1
    -1
      tests/conn_test.py

+ 1
- 1
.gitlab-ci.yml View File

@@ -1,4 +1,4 @@
image: python:3.7-slim
image: python:3.5-slim

before_script:
- python -V


+ 16
- 8
README.md View File

@@ -4,7 +4,7 @@

* Full API coverage
* Https and Mdns discovery
* Thin client wrapper library (~250 lines)
* Thin client wrapper library (<200 lines)
* CLI script example to access any command

[![pipeline status](https://framagit.org/sun/pyfbx/badges/master/pipeline.svg)](https://framagit.org/sun/pyfbx/commits/master)
@@ -17,20 +17,21 @@ A script called pyfbx is installed and can be used like this:
To register the application:

```shell
pyfbx SuperAppId
$ pyfbx SuperAppId
```
Once registration is done on the device, write down the application token.

Using this app token, acquire a session token to execute a test command
Using this app token, acquire a session token to execute a test command.

```shell
pyfbx -t '<TOKEN>' SuperAppId
$ pyfbx -t '<TOKEN>' SuperAppId
$ pyfbx -t 'f:<TOKEN_FILE>' SuperAppId
```

You can also store the token in a file (example token.txt) and also execute a specific command
You can run any commands on the freebox and retrieve complete result or single value

```shell
pyfbx -t f:/home/foo/token.txt SuperAppId -c 'Contacts.Create_a_contact(\
$ pyfbx -t f:/home/foo/token.txt SuperAppId -c 'Contacts.Create_a_contact(\
post_data={"display_name": "Sandy Kilo", "first_name": "Sandy", "last_name":"Kilo"})'
{ 'birthday': '',
'company': '',
@@ -41,13 +42,17 @@ post_data={"display_name": "Sandy Kilo", "first_name": "Sandy", "last_name":"Kil
'last_update': 1554378898,
'notes': '',
'photo_url': ''}

$ pyfbx -t f:token.txt -u https://0s2efr3i.fbxos.fr:15628 id -c "Connection.Get_the_current_Connection_status()['rate_up']"
1700
```

_Note_ : Don't forget to escape token and command with quotes.

The complete script help is:
```shell
pyfbx -h
usage: pyfbx [-h] [-c COMMAND] [-t TOKEN] [-v] [-n] app_id
usage: pyfbx [-h] [-c COMMAND] [-t TOKEN] [-v] [-n] [-u URL] app_id

positional arguments:
app_id application identifier
@@ -61,6 +66,7 @@ optional arguments:
token (or f:<filename>)
-v, --verbose increase verbosity to INFO, use twice for DEBUG
-n, --http Disable MDNS and use http known address
-u URL, --url URL specific url to query
```

## Library usage
@@ -124,4 +130,6 @@ You can run tests with
pip3 install -r requirements-dev.txt
pytest tests
```
Currently to get 100% library coverage, a Freebox sitting on its default IP (192.168.1.254) is required.
The library mechanisms are tested. To get 100% coverage, a Freebox on its default IP (192.168.1.254) is required.

The API description is generated and thus not functionnaly tested.

+ 14
- 5
pyfbx/__main__.py View File

@@ -29,23 +29,32 @@ def main():
parser.add_argument("-v", "--verbose", action="count", default=0,
help="increase verbosity to INFO, use twice for DEBUG")
parser.add_argument("-n", "--http", action="store_true",
help="Disable MDNS and use http known address")
help="disable MDNS and use http known address")
parser.add_argument("-u", "--url", type=str,
help="specific url to query",
default=None)
args = parser.parse_args()
console(log, log_level[min(2, args.verbose)])
try:
myfb = Fbx(nomdns=args.http)
myfb = Fbx(nomdns=args.http, url=args.url)
token = args.token
if token:
if token.startswith('f:'):
with open(args.token[2:]) as tok_file:
token = tok_file.read().strip()
else:
log.warning(f"Registering app {__name__}, id {args.app_id}, Press button")
log.warning("Registering app {}, id {}, Press button".format(__name__, args.app_id))
token = myfb.register(app_id=args.app_id, app_name=__name__,
device=os.uname().nodename)
log.warning(f"Save your application token: {token}")
log.warning("Save your application token: {}".format(token))
myfb.mksession(app_id=args.app_id, token=token)
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(eval(f"myfb.{args.command}"))
pp.pprint(eval("myfb.{}".format(args.command)))
return 0
except BaseException as err:
print(err)
return 2


if __name__ == "__main__":
exit(main())

+ 102
- 102
pyfbx/api.py View File

@@ -9,7 +9,7 @@ _DISC_HTTP_URL = "http://mafreebox.freebox.fr"
_DISC_MDNS_NAME = "_fbx-api._tcp.local."


class _M(namedtuple("Method", ('url', 'doc', 'args'), defaults=([],))):
class _M(namedtuple("Method", ('url', 'doc', 'args'))):
"""
Accessor class for API methods
"""
@@ -30,21 +30,21 @@ class _M(namedtuple("Method", ('url', 'doc', 'args'), defaults=([],))):

SYSTEMS = {
"Download": {
"Retrieve_a_Download_task": _M("GET downloads/", "Retrieve a Download task"),
"Delete_a_Download_task": _M("DELETE downloads/{id}", "Delete a Download task"),
"Retrieve_a_Download_task": _M("GET downloads/", "Retrieve a Download task", []),
"Delete_a_Download_task": _M("DELETE downloads/{id}", "Delete a Download task", []),
"Update_a_Download_task": _M("PUT downloads/{id}", "Update a Download task", ["PostData"]),
"Get_download_log": _M("GET downloads/{id}/log", "Get download log"),
"Get_download_log": _M("GET downloads/{id}/log", "Get download log", []),
"Adding_a_new_Download_task": _M("POST downloads/add", "Adding a new Download task", ["PostData"]),
},
"Download_Feeds": {
"Get_the_list_of_all_download_Feeds": _M("GET downloads/feeds/", "Get the list of all download Feeds"),
"Get_a_download_Feed": _M("GET downloads/feeds/{id}", "Get a download Feed"),
"Get_the_list_of_all_download_Feeds": _M("GET downloads/feeds/", "Get the list of all download Feeds", []),
"Get_a_download_Feed": _M("GET downloads/feeds/{id}", "Get a download Feed", []),
"Add_a_Download_Feed": _M("POST downloads/feeds/", "Add a Download Feed", ["PostData"]),
"Delete_Download_Feed": _M("DELETE downloads/feeds/{id}", "Delete Download Feed"),
"Delete_Download_Feed": _M("DELETE downloads/feeds/{id}", "Delete Download Feed", []),
"Update_a_Download_Feed": _M("PUT downloads/feeds/{id}", "Update a Download Feed", ["PostData"]),
"Refresh_a_Download_Feed": _M("POST downloads/feeds/{id}/fetch", "Refresh a Download Feed", ["PostData"]),
"Refresh_all_Download_Feeds": _M("POST downloads/feeds/fetch", "Refresh all Download Feeds", ["PostData"]),
"Get_the_items_of_a_given_RSS_feed": _M("GET downloads/feeds/{feed_id}/items/", "Get the items of a given RSS feed"),
"Get_the_items_of_a_given_RSS_feed": _M("GET downloads/feeds/{feed_id}/items/", "Get the items of a given RSS feed", []),
"Update_a_feed_item": _M("PUT downloads/feeds/{feed_id}/items/{item_id}", "Update a feed item", ["PostData"]),
"Download_a_feed_item": _M("POST downloads/feeds/{feed_id}/items/{item_id}/download", "Download a feed item", ["PostData"]),
"Mark_all_items_as_read": _M("POST downloads/feeds/{feed_id}/items/mark_all_as_read", "Mark all items as read", ["PostData"]),
@@ -54,12 +54,12 @@ SYSTEMS = {
"Updating_the_current_Throttling_mode": _M("PUT downloads/throttling", "Updating the current Throttling mode", ["PostData"]),
},
"Fs": {
"List_every_tasks": _M("GET fs/tasks/", "List every tasks"),
"List_a_task": _M("GET fs/tasks/{id}", "List a task"),
"Delete_a_task": _M("DELETE fs/tasks/{id}", "Delete a task"),
"List_every_tasks": _M("GET fs/tasks/", "List every tasks", []),
"List_a_task": _M("GET fs/tasks/{id}", "List a task", []),
"Delete_a_task": _M("DELETE fs/tasks/{id}", "Delete a task", []),
"Update_a_task": _M("PUT fs/tasks/{id}", "Update a task", ["PostData"]),
"List_files": _M("GET fs/ls/{path}", "List files"),
"Get_file_information": _M("GET fs/info/{path}", "Get file information"),
"List_files": _M("GET fs/ls/{path}", "List files", []),
"Get_file_information": _M("GET fs/info/{path}", "Get file information", []),
"Conflict_resolution": _M("POST fs/mv/", "Conflict resolution", ["PostData"]),
"Move_files": _M("POST fs/mv/", "Move files", ["PostData"]),
"Copy_files": _M("POST fs/cp/", "Copy files", ["PostData"]),
@@ -70,185 +70,185 @@ SYSTEMS = {
"Repair_a_file": _M("POST fs/repair/", "Repair a file", ["PostData"]),
"Hash_a_file": _M("POST fs/hash/", "Hash a file", ["PostData"]),
"Create_a_directory": _M("POST fs/mkdir/", "Create a directory", ["PostData"]),
"Rename_a_file/folder": _M("POST fs/rename/", "Rename a file/folder", ["PostData"]),
"Download_a_file": _M("GET dl/{path}", "Download a file"),
"Rename_a_file_folder": _M("POST fs/rename/", "Rename a file/folder", ["PostData"]),
"Download_a_file": _M("GET dl/{path}", "Download a file", []),
},
"Share": {
"Retrieve_a_File_Sharing_link": _M("GET share_link/", "Retrieve a File Sharing link"),
"Delete_a_File_Sharing_link": _M("DELETE share_link/{token}", "Delete a File Sharing link"),
"Retrieve_a_File_Sharing_link": _M("GET share_link/", "Retrieve a File Sharing link", []),
"Delete_a_File_Sharing_link": _M("DELETE share_link/{token}", "Delete a File Sharing link", []),
"Create_a_File_Sharing_link": _M("POST share_link/", "Create a File Sharing link", ["PostData"]),
},
"Upload": {
"File_Upload_example": _M("GET ws/upload", "File Upload example"),
"Get_the_list_of_uploads": _M("GET upload/", "Get the list of uploads"),
"Track_an_upload_status": _M("GET upload/{id}", "Track an upload status"),
"Cancel_an_upload": _M("DELETE upload/{id}/cancel", "Cancel an upload"),
"Delete_an_upload": _M("DELETE upload/{id}", "Delete an upload"),
"Cleanup_all_terminated_uploads": _M("DELETE upload/clean", "Cleanup all terminated uploads"),
"File_Upload_example": _M("GET ws/upload", "File Upload example", []),
"Get_the_list_of_uploads": _M("GET upload/", "Get the list of uploads", []),
"Track_an_upload_status": _M("GET upload/{id}", "Track an upload status", []),
"Cancel_an_upload": _M("DELETE upload/{id}/cancel", "Cancel an upload", []),
"Delete_an_upload": _M("DELETE upload/{id}", "Delete an upload", []),
"Cleanup_all_terminated_uploads": _M("DELETE upload/clean", "Cleanup all terminated uploads", []),
},
"Airmedia": {
"Get_the_current_AirMedia_configuration": _M("GET airmedia/config/", "Get the current AirMedia configuration"),
"Get_the_current_AirMedia_configuration": _M("GET airmedia/config/", "Get the current AirMedia configuration", []),
"Update_the_current_AirMedia_configuration": _M("PUT airmedia/config/", "Update the current AirMedia configuration", ["PostData"]),
"Get_the_list_of_available_AirMedia_receivers": _M("GET airmedia/receivers/", "Get the list of available AirMedia receivers"),
"Get_the_list_of_available_AirMedia_receivers": _M("GET airmedia/receivers/", "Get the list of available AirMedia receivers", []),
},
"Rrd": {
},
"Call": {
"List_every_calls": _M("GET call/log/", "List every calls"),
"List_every_calls": _M("GET call/log/", "List every calls", []),
"Delete_every_calls": _M("POST call/log/delete_all/", "Delete every calls", ["PostData"]),
"Mark_every_calls_as_read": _M("POST call/log/mark_all_as_read/", "Mark every calls as read", ["PostData"]),
"Access_a_given_call_entry": _M("GET call/log/{id}", "Access a given call entry"),
"Delete_a_call": _M("DELETE call/log/{id}", "Delete a call"),
"Access_a_given_call_entry": _M("GET call/log/{id}", "Access a given call entry", []),
"Delete_a_call": _M("DELETE call/log/{id}", "Delete a call", []),
"Update_a_call_entry": _M("PUT call/log/{id}", "Update a call entry", ["PostData"]),
},
"Contacts": {
"Get_a_list_of_contacts": _M("GET contact/", "Get a list of contacts"),
"Access_a_given_contact_entry": _M("GET contact/{id}", "Access a given contact entry"),
"Get_a_list_of_contacts": _M("GET contact/", "Get a list of contacts", []),
"Access_a_given_contact_entry": _M("GET contact/{id}", "Access a given contact entry", []),
"Create_a_contact": _M("POST contact/", "Create a contact", ["PostData"]),
"Delete_a_contact": _M("DELETE contact/{id}", "Delete a contact"),
"Delete_a_contact": _M("DELETE contact/{id}", "Delete a contact", []),
"Update_a_contact_entry": _M("PUT contact/{id}", "Update a contact entry", ["PostData"]),
"Get_the_list_of_numbers_for_a_given_contact": _M("GET contact/{contact_id}/[numbers|addresses|urls|emails]/", "Get the list of numbers for a given contact"),
"Access_a_given_contact_number": _M("GET [number,address,url,email]/{id}", "Access a given contact number"),
"Get_the_list_of_numbers_for_a_given_contact": _M("GET contact/{contact_id}/[numbers|addresses|urls|emails]/", "Get the list of numbers for a given contact", []),
"Access_a_given_contact_number": _M("GET [number,address,url,email]/{id}", "Access a given contact number", []),
"Create_a_contact_number": _M("POST [number,address,url,email]/", "Create a contact number", ["PostData"]),
"Delete_a_contact_number": _M("DELETE [number,address,url,email]/{id}", "Delete a contact number"),
"Delete_a_contact_number": _M("DELETE [number,address,url,email]/{id}", "Delete a contact number", []),
"Update_a_contact_number": _M("PUT [number,address,url,email]/{id}", "Update a contact number", ["PostData"]),
},
"Connection": {
"Get_the_current_Connection_status": _M("GET connection/", "Get the current Connection status"),
"Get_the_current_Connection_configuration": _M("GET connection/config/", "Get the current Connection configuration"),
"Get_the_current_Connection_status": _M("GET connection/", "Get the current Connection status", []),
"Get_the_current_Connection_configuration": _M("GET connection/config/", "Get the current Connection configuration", []),
"Update_the_Connection_configuration": _M("PUT connection/config/", "Update the Connection configuration", ["PostData"]),
"Get_the_current_IPv6_Connection_configuration": _M("GET connection/ipv6/config/", "Get the current IPv6 Connection configuration"),
"Get_the_current_IPv6_Connection_configuration": _M("GET connection/ipv6/config/", "Get the current IPv6 Connection configuration", []),
"Update_the_IPv6_Connection_configuration": _M("PUT connection/ipv6/config/", "Update the IPv6 Connection configuration", ["PostData"]),
"Get_the_current_xDSL_infos": _M("GET connection/xdsl/", "Get the current xDSL infos"),
"Get_the_current_FTTH_status": _M("GET connection/ftth/", "Get the current FTTH status"),
"Get_the_status_of_a_DynDNS_service": _M("GET connection/ddns/{provider}/status/", "Get the status of a DynDNS service"),
"Get_the_config_of_a_DynDNS_service": _M("GET connection/ddns/{provider}/", "Get the config of a DynDNS service"),
"Get_the_current_xDSL_infos": _M("GET connection/xdsl/", "Get the current xDSL infos", []),
"Get_the_current_FTTH_status": _M("GET connection/ftth/", "Get the current FTTH status", []),
"Get_the_status_of_a_DynDNS_service": _M("GET connection/ddns/{provider}/status/", "Get the status of a DynDNS service", []),
"Get_the_config_of_a_DynDNS_service": _M("GET connection/ddns/{provider}/", "Get the config of a DynDNS service", []),
"Set_the_config_of_a_DynDNS_service": _M("PUT connection/ddns/{provider}/", "Set the config of a DynDNS service", ["PostData"]),
},
"Lan": {
"Get_the_current_Lan_configuration": _M("GET lan/config/", "Get the current Lan configuration"),
"Get_the_current_Lan_configuration": _M("GET lan/config/", "Get the current Lan configuration", []),
"Update_the_current_Lan_configuration": _M("PUT lan/config/", "Update the current Lan configuration", ["PostData"]),
},
"Freeplug": {
"Get_the_current_Freeplugs_networks": _M("GET freeplug/", "Get the current Freeplugs networks"),
"Get_a_particular_Freeplug_information": _M("GET freeplug/{id}/", "Get a particular Freeplug information"),
"Reset_a_Freeplug": _M("POST freeplug/{id}/reset/", "Reset a Freeplug"),
"Get_the_current_Freeplugs_networks": _M("GET freeplug/", "Get the current Freeplugs networks", []),
"Get_a_particular_Freeplug_information": _M("GET freeplug/{id}/", "Get a particular Freeplug information", []),
"Reset_a_Freeplug": _M("POST freeplug/{id}/reset/", "Reset a Freeplug", []),
},
"Dhcp": {
"Get_the_current_DHCP_configuration": _M("GET dhcp/config/", "Get the current DHCP configuration"),
"Get_the_current_DHCP_configuration": _M("GET dhcp/config/", "Get the current DHCP configuration", []),
"Update_the_current_DHCP_configuration": _M("PUT dhcp/config/", "Update the current DHCP configuration", ["PostData"]),
"Get_the_list_of_DHCP_static_leases": _M("GET dhcp/static_lease/", "Get the list of DHCP static leases"),
"Get_a_given_DHCP_static_lease": _M("GET dhcp/static_lease/{id}", "Get a given DHCP static lease"),
"Get_the_list_of_DHCP_static_leases": _M("GET dhcp/static_lease/", "Get the list of DHCP static leases", []),
"Get_a_given_DHCP_static_lease": _M("GET dhcp/static_lease/{id}", "Get a given DHCP static lease", []),
"Update_DHCP_static_lease": _M("PUT dhcp/static_lease/{id}", "Update DHCP static lease", ["PostData"]),
"Delete_a_DHCP_static_lease": _M("DELETE dhcp/static_lease/{id}", "Delete a DHCP static lease"),
"Delete_a_DHCP_static_lease": _M("DELETE dhcp/static_lease/{id}", "Delete a DHCP static lease", []),
"Add_a_DHCP_static_lease": _M("POST dhcp/static_lease/", "Add a DHCP static lease", ["PostData"]),
"Get_the_list_of_DHCP_dynamic_leases": _M("GET dhcp/dynamic_lease/", "Get the list of DHCP dynamic leases"),
"Get_the_list_of_DHCP_dynamic_leases": _M("GET dhcp/dynamic_lease/", "Get the list of DHCP dynamic leases", []),
},
"Ftp": {
"Get_the_current_Ftp_configuration": _M("GET ftp/config/", "Get the current Ftp configuration"),
"Get_the_current_Ftp_configuration": _M("GET ftp/config/", "Get the current Ftp configuration", []),
"Update_the_FTP_configuration": _M("PUT ftp/config/", "Update the FTP configuration", ["PostData"]),
},
"Nat": {
"Get_the_current_Dmz_configuration": _M("GET fw/dmz/", "Get the current Dmz configuration"),
"Get_the_current_Dmz_configuration": _M("GET fw/dmz/", "Get the current Dmz configuration", []),
"Update_the_current_Dmz_configuration": _M("PUT fw/dmz/", "Update the current Dmz configuration", ["PostData"]),
},
"Igd": {
"Get_the_current_UPnP_IGD_configuration": _M("GET upnpigd/config/", "Get the current UPnP IGD configuration"),
"Get_the_current_UPnP_IGD_configuration": _M("GET upnpigd/config/", "Get the current UPnP IGD configuration", []),
"Update_the_UPnP_IGD_configuration": _M("PUT upnpigd/config/", "Update the UPnP IGD configuration", ["PostData"]),
"Get_the_list_of_current_redirection": _M("GET upnpigd/redir/", "Get the list of current redirection"),
"Delete_a_redirection": _M("DELETE upnpigd/redir/{id}", "Delete a redirection"),
"Get_the_list_of_current_redirection": _M("GET upnpigd/redir/", "Get the list of current redirection", []),
"Delete_a_redirection": _M("DELETE upnpigd/redir/{id}", "Delete a redirection", []),
},
"Lcd": {
"Get_the_current_LCD_configuration": _M("GET lcd/config/", "Get the current LCD configuration"),
"Get_the_current_LCD_configuration": _M("GET lcd/config/", "Get the current LCD configuration", []),
"Update_the_lcd_configuration": _M("PUT lcd/config/", "Update the lcd configuration", ["PostData"]),
},
"Network_Share": {
"Get_the_current_Samba_configuration": _M("GET netshare/samba/", "Get the current Samba configuration"),
"Get_the_current_Samba_configuration": _M("GET netshare/samba/", "Get the current Samba configuration", []),
"Update_the_Samba_configuration": _M("PUT netshare/samba/", "Update the Samba configuration", ["PostData"]),
"Get_the_current_Afp_configuration": _M("GET netshare/afp/", "Get the current Afp configuration"),
"Get_the_current_Afp_configuration": _M("GET netshare/afp/", "Get the current Afp configuration", []),
"Update_the_Afp_configuration": _M("PUT netshare/afp/", "Update the Afp configuration", ["PostData"]),
},
"Upnpav": {
"Get_the_current_UPnP_AV_configuration": _M("GET upnpav/config/", "Get the current UPnP AV configuration"),
"Get_the_current_UPnP_AV_configuration": _M("GET upnpav/config/", "Get the current UPnP AV configuration", []),
"Update_the_UPnP_AV_configuration": _M("PUT upnpav/config/", "Update the UPnP AV configuration", ["PostData"]),
},
"Switch": {
"Get_the_current_switch_status": _M("GET switch/status/", "Get the current switch status"),
"Get_a_port_configuration": _M("GET switch/port/{id}", "Get a port configuration"),
"Get_the_current_switch_status": _M("GET switch/status/", "Get the current switch status", []),
"Get_a_port_configuration": _M("GET switch/port/{id}", "Get a port configuration", []),
"Update_a_port_configuration": _M("PUT switch/port/{id}", "Update a port configuration", ["PostData"]),
"Get_a_port_stats": _M("GET switch/port/{id}/stats", "Get a port stats"),
"Get_a_port_stats": _M("GET switch/port/{id}/stats", "Get a port stats", []),
},
"Wifi": {
"Get_the_current_Wi-Fi_global_configuration": _M("GET wifi/config/", "Get the current Wi-Fi global configuration"),
"Update_the_Wi-Fi_global_configuration": _M("PUT wifi/config/", "Update the Wi-Fi global configuration", ["PostData"]),
"Wi-Fi_AP_API": _M("GET wifi/ap/", "Wi-Fi AP API"),
"Get_Wi-Fi_Stations_List": _M("GET wifi/ap/{id}/stations/", "Get Wi-Fi Stations List"),
"Wi-Fi_BSS_API": _M("GET wifi/bss/", "Wi-Fi BSS API"),
"List_AP_neighbors": _M("GET wifi/ap/{id}/neighbors/", "List AP neighbors"),
"List_Wi-Fi_channels_usage": _M("GET wifi/ap/{id}/channel_usage/", "List Wi-Fi channels usage"),
"Get_the_current_WiFi_global_configuration": _M("GET wifi/config/", "Get the current Wi-Fi global configuration", []),
"Update_the_WiFi_global_configuration": _M("PUT wifi/config/", "Update the Wi-Fi global configuration", ["PostData"]),
"WiFi_AP_API": _M("GET wifi/ap/", "Wi-Fi AP API", []),
"Get_WiFi_Stations_List": _M("GET wifi/ap/{id}/stations/", "Get Wi-Fi Stations List", []),
"WiFi_BSS_API": _M("GET wifi/bss/", "Wi-Fi BSS API", []),
"List_AP_neighbors": _M("GET wifi/ap/{id}/neighbors/", "List AP neighbors", []),
"List_WiFi_channels_usage": _M("GET wifi/ap/{id}/channel_usage/", "List Wi-Fi channels usage", []),
"Refresh_radar_informations": _M("POST wifi/ap/{id}/neighbors/scan", "Refresh radar informations", ["PostData"]),
"Get_Wi-Fi_Planning": _M("GET wifi/planning/", "Get Wi-Fi Planning"),
"Update_Wi-Fi_Planning": _M("PUT wifi/planning/", "Update Wi-Fi Planning", ["PostData"]),
"Get_the_MAC_filter_list": _M("GET wifi/mac_filter/", "Get the MAC filter list"),
"Getting_a_particular_MAC_filter": _M("GET wifi/mac_filter/{filter_id}", "Getting a particular MAC filter"),
"Get_WiFi_Planning": _M("GET wifi/planning/", "Get Wi-Fi Planning", []),
"Update_WiFi_Planning": _M("PUT wifi/planning/", "Update Wi-Fi Planning", ["PostData"]),
"Get_the_MAC_filter_list": _M("GET wifi/mac_filter/", "Get the MAC filter list", []),
"Getting_a_particular_MAC_filter": _M("GET wifi/mac_filter/{filter_id}", "Getting a particular MAC filter", []),
"Updating_a_MAC_filter": _M("PUT wifi/mac_filter/{filter_id}", "Updating a MAC filter", ["PostData"]),
"Delete_a_MAC_filter": _M("DELETE wifi/mac_filter/{filter_id}", "Delete a MAC filter"),
"Delete_a_MAC_filter": _M("DELETE wifi/mac_filter/{filter_id}", "Delete a MAC filter", []),
"Create_a_new_MAC_filter": _M("POST wifi/mac_filter/", "Create a new MAC filter", ["PostData"]),
},
"System": {
"Get_the_current_system_info": _M("GET system/", "Get the current system info [UNSTABLE]"),
"Reboot_the_system": _M("POST system/reboot/", "Reboot the system"),
"Get_the_current_system_info": _M("GET system/", "Get the current system info [UNSTABLE]", []),
"Reboot_the_system": _M("POST system/reboot/", "Reboot the system", []),
},
"Vpn": {
"VPN_Server_List_API": _M("GET vpn/", "VPN Server List API"),
"Get_a_VPN_config": _M("GET vpn/{vpn_id}/config/", "Get a VPN config"),
"VPN_Server_List_API": _M("GET vpn/", "VPN Server List API", []),
"Get_a_VPN_config": _M("GET vpn/{vpn_id}/config/", "Get a VPN config", []),
"Update_the_VPN_configuration": _M("PUT vpn/openvpn_routed/config/", "Update the VPN configuration", ["PostData"]),
"VPN_Server_User_List": _M("GET vpn/user/", "VPN Server User List"),
"Get_a_VPN_user": _M("GET vpn/user/{login}", "Get a VPN user"),
"VPN_Server_User_List": _M("GET vpn/user/", "VPN Server User List", []),
"Get_a_VPN_user": _M("GET vpn/user/{login}", "Get a VPN user", []),
"Add_a_VPN_User": _M("POST vpn/user/", "Add a VPN User", ["PostData"]),
"Delete_a_VPN_User": _M("DELETE vpn/user/{login}", "Delete a VPN User"),
"Delete_a_VPN_User": _M("DELETE vpn/user/{login}", "Delete a VPN User", []),
"Update_a_VPN_User": _M("PUT vpn/user/{login}", "Update a VPN User", ["PostData"]),
"Get_the_VPN_server_IP_pool_reservations": _M("GET vpn/ip_pool/", "Get the VPN server IP pool reservations"),
"Get_the_list_of_connections": _M("GET vpn/connection/", "Get the list of connections"),
"Close_a_given_connection": _M("DELETE vpn/connection/{id}", "Close a given connection"),
"Donwload_a_user_configuration_file": _M("GET vpn/download_config/{server_name}/{login}", "Donwload a user configuration file"),
"Get_the_VPN_server_IP_pool_reservations": _M("GET vpn/ip_pool/", "Get the VPN server IP pool reservations", []),
"Get_the_list_of_connections": _M("GET vpn/connection/", "Get the list of connections", []),
"Close_a_given_connection": _M("DELETE vpn/connection/{id}", "Close a given connection", []),
"Donwload_a_user_configuration_file": _M("GET vpn/download_config/{server_name}/{login}", "Donwload a user configuration file", []),
},
"Vpn_Client": {
"Get_VPN_Client_configuration_list": _M("GET vpn_client/config/", "Get VPN Client configuration list"),
"Get_a_VPN_client_config": _M("GET vpn_client/config/{id}", "Get a VPN client config"),
"Get_VPN_Client_configuration_list": _M("GET vpn_client/config/", "Get VPN Client configuration list", []),
"Get_a_VPN_client_config": _M("GET vpn_client/config/{id}", "Get a VPN client config", []),
"Add_a_VPN_client_configuration": _M("POST vpn_client/config/", "Add a VPN client configuration", ["PostData"]),
"Delete_a_VPN_client_Configuration": _M("DELETE vpn_client/config/{id}", "Delete a VPN client Configuration"),
"Delete_a_VPN_client_Configuration": _M("DELETE vpn_client/config/{id}", "Delete a VPN client Configuration", []),
"Update_the_VPN_client_configuration": _M("PUT vpn_client/config/{id}", "Update the VPN client configuration", ["PostData"]),
"Get_the_VPN_client_status": _M("GET vpn_client/status", "Get the VPN client status"),
"Get_the_VPN_client_logs": _M("GET vpn_client/log", "Get the VPN client logs"),
"Get_the_VPN_client_status": _M("GET vpn_client/status", "Get the VPN client status", []),
"Get_the_VPN_client_logs": _M("GET vpn_client/log", "Get the VPN client logs", []),
},
"Storage": {
"Get_the_list_of_disks": _M("GET storage/disk/", "Get the list of disks"),
"Get_a_given_disk_info": _M("GET storage/disk/{id}", "Get a given disk info"),
"Get_the_list_of_disks": _M("GET storage/disk/", "Get the list of disks", []),
"Get_a_given_disk_info": _M("GET storage/disk/{id}", "Get a given disk info", []),
"Update_a_disk_state": _M("PUT storage/disk/{id}", "Update a disk state", ["PostData"]),
"Format_a_disk": _M("PUT storage/disk/{id}/format/", "Format a disk", ["PostData"]),
"Get_the_list_of_partitions": _M("GET storage/partition/", "Get the list of partitions"),
"Get_a_given_partition_info": _M("GET storage/partition/{id}", "Get a given partition info"),
"Get_the_list_of_partitions": _M("GET storage/partition/", "Get the list of partitions", []),
"Get_a_given_partition_info": _M("GET storage/partition/{id}", "Get a given partition info", []),
"Update_a_partition_state": _M("PUT storage/partition/{id}", "Update a partition state", ["PostData"]),
"Check_a_partition": _M("PUT storage/partition/{id}/check/", "Check a partition", ["PostData"]),
"Get_the_current_storage_configuration": _M("GET storage/config/", "Get the current storage configuration"),
"Get_the_current_storage_configuration": _M("GET storage/config/", "Get the current storage configuration", []),
"Update_the_External_Storage_configuration": _M("PUT storage/config/", "Update the External Storage configuration", ["PostData"]),
},
"Parental": {
"Get_parental_filter_config": _M("GET parental/config/", "Get parental filter config"),
"Get_parental_filter_config": _M("GET parental/config/", "Get parental filter config", []),
"Update_parental_filter_config": _M("PUT parental/config/", "Update parental filter config", ["PostData"]),
"Retrieve_all_Parental_Filter_rules": _M("GET parental/filter/", "Retrieve all Parental Filter rules"),
"Retrieve_a_Parental_Filter_rule": _M("GET parental/filter/{id}", "Retrieve a Parental Filter rule"),
"Delete_a_Parental_Filter_rule": _M("DELETE parental/filter/{id}", "Delete a Parental Filter rule"),
"Retrieve_all_Parental_Filter_rules": _M("GET parental/filter/", "Retrieve all Parental Filter rules", []),
"Retrieve_a_Parental_Filter_rule": _M("GET parental/filter/{id}", "Retrieve a Parental Filter rule", []),
"Delete_a_Parental_Filter_rule": _M("DELETE parental/filter/{id}", "Delete a Parental Filter rule", []),
"Update_a_Parental_Filter_rule": _M("PUT parental/filter/{id}", "Update a Parental Filter rule", ["PostData"]),
"Add_a_Parental_Filter_rule": _M("POST parental/filter/", "Add a Parental Filter rule", ["PostData"]),
"Get_a_parental_filter_Planning": _M("GET parental/filter/{id}/planning", "Get a parental filter Planning"),
"Get_a_parental_filter_Planning": _M("GET parental/filter/{id}/planning", "Get a parental filter Planning", []),
"Update_a_parental_filter_Planning": _M("PUT parental/filter/{id}/planning", "Update a parental filter Planning", ["PostData"]),
},
"Pvr": {
"Get_the_current_PVR_configuration": _M("GET pvr/config/", "Get the current PVR configuration"),
"Get_the_current_PVR_configuration": _M("GET pvr/config/", "Get the current PVR configuration", []),
"Update_the_current_PVR_configuration": _M("PUT pvr/config/", "Update the current PVR configuration", ["PostData"]),
"Getting_the_current_quota_info": _M("GET pvr/quota/", "Getting the current quota info"),
"Getting_the_current_quota_info": _M("GET pvr/quota/", "Getting the current quota info", []),
"Request_next_quota_threshold": _M("PUT pvr/quota/", "Request next quota threshold", ["PostData"]),
},
}

+ 8
- 9
pyfbx/client.py View File

@@ -7,7 +7,6 @@ import logging
import re
import hmac
import hashlib
import logging
import time
import requests
from . import api
@@ -67,14 +66,14 @@ class FbxTransport():
response = req_response.json()
if response['success']:
if 'result' in response:
self.log.debug(f"<< Got {response['result']}")
self.log.debug("<< Got {}".format(response['result']))
return response['result']
else:
raise FbxErrorResponse(response['error_code'], response['msg'])

def get_local_base(self, url=api._DISC_HTTP_URL):
response = self._session.get(f"{url}/api_version").json()
self.log.debug(f"<< Detected api {response['api_version']}")
response = self._session.get("{}/api_version".format(url)).json()
self.log.debug("<< Detected api {}".format(response['api_version']))
return "%s%sv%s" % (url, response['api_base_url'],
response['api_version'][0])

@@ -106,10 +105,10 @@ class Fbx():
s = "pending"
self.log.info("Press Ok on the freebox to register application")
while s == "pending":
s = self._trn.api_exec("GET", f"login/authorize/{trackid}")["status"]
s = self._trn.api_exec("GET", "login/authorize/{}".format(trackid))["status"]
if s == "pending":
time.sleep(1)
self.log.debug(f"Registration returned: {s}")
self.log.debug("Registration returned: {}".format(s))
return s == "granted" and self.token

def mksession(self, app_id=None, token=None):
@@ -129,7 +128,7 @@ class Fbx():
session_token = resp["session_token"]
self.app_id = app_id
self._trn.set_session_header(session_token)
self.log.info(f"Authenticated")
self.log.info("Authenticated".format())
return resp["permissions"]


@@ -139,13 +138,13 @@ class FbxErrorResponse(Exception):
self.msg = msg

def __str__(self):
return f'{self.msg} [{self.error_code}]'
return '{} [{}]'.format(self.msg, self.error_code)


log = logging.getLogger("pyfbx")

# All FB subsystems are classes deriving from FbxClass
for _classname in api.SYSTEMS:
log.debug(f"Adding class {_classname} to locals")
log.debug("Adding class {} to locals".format(_classname))
locals()[_classname] = type(_classname, (FbxClass, ), {})
__all__.append(_classname)

+ 2
- 2
pyfbx/mdns.py View File

@@ -45,9 +45,9 @@ class FbxMDNS():
prop = self.svc_prop
base = "%sv%s" % (prop['api_base_url'], prop['api_version'][0])
if int(prop['https_available']):
return f"https://{prop['api_domain']}:{prop['https_port']}{base}"
return "https://{}:{}{}".format(prop['api_domain'], prop['https_port'], base)
else:
return f"http://{prop['api_domain']}{base}"
return "http://{}{}".format(prop['api_domain'], base)

@property
def svc_info(self):


+ 4
- 4
pyfbx/utils.py View File

@@ -37,11 +37,11 @@ def add_class_func(cls, name, method):

params = (["post_data"] if method.args else []) + get_params(method.endpoint)
input_params = ", ".join(params)
passed_param = ", ".join([f"{e}={e}" for e in params])
passed_param = ", ".join(["{}={}".format(e, e) for e in params])
namespace = {}
code = (f"def func(self, {input_params}):\n"
f" return self._trn.api_exec(\n"
f" \"{method.http_method}\", \"{method.endpoint}\", {passed_param})")
code = ("def func(self, {}):\n"
" return self._trn.api_exec(\n"
" \"{}\", \"{}\", {})".format(input_params, method.http_method, method.endpoint, passed_param))
exec(code, namespace)
func = namespace['func']
func.__name__ = name


+ 1
- 1
setup.py View File

@@ -3,7 +3,7 @@ from setuptools import setup, find_packages

setup(
name="pyfbx",
version="0.0.4",
version="0.0.5",
description="Freebox thin client",
long_description=(pathlib.Path(__file__).parent / "README.md").read_text(),
long_description_content_type='text/markdown',


+ 1
- 1
tests/conn_test.py View File

@@ -29,7 +29,7 @@ def test_fbx_session_local():
with open(tokname) as tokfile:
token = tokfile.readline().rstrip()
app_id = tokfile.readline().rstrip()
print(f"token:{token}, app_id:{app_id}")
print("token:{}, app_id:{}".format(token, app_id))
f = Fbx()
f.mksession(app_id=app_id, token=token)
r = f.Contacts.Create_a_contact(post_data={"display_name": "Sandy Kilo",


Loading…
Cancel
Save