nas/synology-api-python
1
0
Fork 0
mirror of https://github.com/ProtoThis/python-synology.git synced 2026-01-16 15:09:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add tests + some fixes (#23)

Browse Source 
* Add tests + some fixes

Fixes:
- [utilization] cpu_total_load returns `None` if one load is 0
- [storage] volume_disk_temp_avg + volume_disk_temp_max were global, not volume related

Add:
- [storage] env
- [storage] storage_pools

Changes:
- api.password is now private
- api.storage_url is now private
- use API_KEY on each of APIs/services

* Fix encoding

* Use future division

* Add test_disk + test_volume to check error handling cases
This commit is contained in:
Quentame
2020-03-30 23:07:54 +02:00
committed by GitHub
parent b80b1f3cae
commit aa9bba8814
12 changed files with 813 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.travis.yml
View File

@ -9,15 +9,17 @@ python:
cache:
pip: true
install:
before_install:
- pip install -r requirements_all.txt
- pip install -e .
before_script:
- pylint synology_dsm
- ./scripts/check_format.sh
script:
install:
- python setup.py install
- python setup.py sdist
before_script:
- pylint synology_dsm tests
- ./scripts/check_format.sh
script:
- py.test
deploy:
- provider: pypi

1
requirements_test.txt
View File

@ -1,2 +1,3 @@
pytest
pylint>=1.9.5,<=2.4.4
pylint-strict-informational==0.1

3
setup.cfg Normal file
View File

@ -0,0 +1,3 @@
[tool:pytest]
testpaths = tests
norecursedirs=.git .tox build lib

2
setup.py
View File

@ -1,5 +1,5 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# -*- coding: utf-8 -*-
# NOTE(ProtoThis) Guidelines for Major.Minor.Micro
# - Major means an API contract change

6
synology_dsm/api/core/utilization.py
View File

@ -1,11 +1,13 @@
# -*- coding: utf-8 -*-
"""DSM Utilization data."""
# -*- coding:utf-8 -*-
from synology_dsm.helpers import SynoFormatHelper
class SynoCoreUtilization(object):
"""Class containing Utilization data."""
API_KEY = "SYNO.Core.System.Utilization"
def __init__(self, raw_data):
self._data = {}
self.update(raw_data)
@ -42,7 +44,7 @@ class SynoCoreUtilization(object):
user_load = self.cpu_user_load
other_load = self.cpu_other_load
if system_load and user_load and other_load:
if system_load is not None and user_load is not None and other_load is not None:
return system_load + user_load + other_load
return None

4
synology_dsm/api/dsm/information.py
View File

@ -1,10 +1,12 @@
# -*- coding: utf-8 -*-
"""DSM Information data."""
# -*- coding:utf-8 -*-
class SynoDSMInformation(object):
"""Class containing Information data."""
API_KEY = "SYNO.DSM.Info"
def __init__(self, raw_data):
self._data = {}
self.update(raw_data)

81
synology_dsm/api/storage/storage.py
View File

@ -1,11 +1,15 @@
# -*- coding: utf-8 -*-
"""DSM Storage data."""
# -*- coding:utf-8 -*-
from __future__ import division
from synology_dsm.helpers import SynoFormatHelper
class SynoStorage(object):
"""Class containing Storage data."""
API_KEY = "SYNO.Storage.CGI.Storage"
def __init__(self, raw_data):
self._data = {}
self.update(raw_data)
@ -15,11 +19,28 @@ class SynoStorage(object):
if raw_data:
self._data = raw_data["data"]
# Root
@property
def disks(self):
"""Gets all (internal) disks."""
return self._data.get("disks", [])
@property
def env(self):
"""Gets storage env."""
return self._data.get("env")
@property
def storage_pools(self):
"""Gets all storage pools."""
return self._data.get("storagePools", [])
@property
def volumes(self):
"""Gets all volumes."""
return self._data.get("volumes", [])
# Volume
@property
def volumes_ids(self):
"""Returns volumes ids."""
@ -76,43 +97,35 @@ class SynoStorage(object):
def volume_disk_temp_avg(self, volume_id):
"""Average temperature of all disks making up the volume."""
volume = self._get_volume(volume_id)
if volume:
vol_disks = self.disks
if vol_disks:
total_temp = 0
total_disks = 0
vol_disks = self._get_disks_for_volume(volume_id)
if vol_disks:
total_temp = 0
total_disks = 0
for vol_disk in vol_disks:
disk_temp = self.disk_temp(vol_disk)
if disk_temp:
total_disks += 1
total_temp += disk_temp
for vol_disk in vol_disks:
disk_temp = self.disk_temp(vol_disk["id"])
if disk_temp:
total_disks += 1
total_temp += disk_temp
if total_temp > 0 and total_disks > 0:
return round(total_temp / total_disks, 0)
if total_temp > 0 and total_disks > 0:
return round(total_temp / total_disks, 0)
return None
def volume_disk_temp_max(self, volume_id):
"""Maximum temperature of all disks making up the volume."""
volume = self._get_volume(volume_id)
if volume:
vol_disks = self.disks
if vol_disks:
max_temp = 0
vol_disks = self._get_disks_for_volume(volume_id)
if vol_disks:
max_temp = 0
for vol_disk in vol_disks:
disk_temp = self.disk_temp(vol_disk)
if disk_temp and disk_temp > max_temp:
max_temp = disk_temp
return max_temp
for vol_disk in vol_disks:
disk_temp = self.disk_temp(vol_disk["id"])
if disk_temp and disk_temp > max_temp:
max_temp = disk_temp
return max_temp
return None
@property
def disks(self):
"""Gets all (internal) disks."""
return self._data.get("disks", [])
# Disk
@property
def disks_ids(self):
"""Returns (internal) disks ids."""
@ -128,6 +141,16 @@ class SynoStorage(object):
return disk
return {}
def _get_disks_for_volume(self, volume_id):
"""Returns a list of disk for a specific volume."""
disks = []
pools = self._data.get("storagePools", [])
for pool in pools:
if pool["deploy_path"] == volume_id:
for disk_id in pool["disks"]:
disks.append(self._get_disk(disk_id))
return disks
def disk_name(self, disk_id):
"""The name of this disk."""
return self._get_disk(disk_id).get("name")

2
synology_dsm/helpers.py
View File

@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
"""Helpers."""
# -*- coding:utf-8 -*-
class SynoFormatHelper(object):

41
synology_dsm/synology_dsm.py
View File

@ -1,5 +1,5 @@
"""Classe to interact with Synology DSM."""
# -*- coding:utf-8 -*-
# -*- coding: utf-8 -*-
"""Class to interact with Synology DSM."""
import urllib3
import requests
from requests.compat import json
@ -30,7 +30,7 @@ class SynologyDSM(object):
):
# Store Variables
self.username = username
self.password = password
self._password = password
# Class Variables
self.access_token = None
@ -59,12 +59,12 @@ class SynologyDSM(object):
if self._dsm_version == 5:
if self._use_https:
self.storage_url = (
self._storage_url = (
"https://%s:%s/webman/modules/StorageManager/storagehandler.cgi"
% (dsm_ip, dsm_port)
)
else:
self.storage_url = (
self._storage_url = (
"http://%s:%s/webman/modules/StorageManager/storagehandler.cgi"
% (dsm_ip, dsm_port)
)
@ -79,7 +79,7 @@ class SynologyDSM(object):
# encoding special characters
auth = {
"account": self.username,
"passwd": self.password,
"passwd": self._password,
}
return urlencode(auth)
@ -92,9 +92,9 @@ class SynologyDSM(object):
self._session = requests.Session()
self._session.verify = False
api_path = "%s/auth.cgi?api=SYNO.API.Auth&version=2" % (self.base_url,)
api_path = "%s/auth.cgi?api=SYNO.API.Auth&version=2" % self.base_url
login_path = "method=login&%s" % (self._encode_credentials())
login_path = "method=login&%s" % self._encode_credentials()
url = "%s&%s&session=Core&format=cookie" % (api_path, login_path)
result = self._execute_get_url(url, False)
@ -168,38 +168,35 @@ class SynologyDSM(object):
def update(self, with_information=False):
"""Updates the various instanced modules."""
if self._information and with_information:
api = "SYNO.DSM.Info"
version = 1
if self._dsm_version >= 6:
version = 2
url = "%s/entry.cgi?api=%s&version=%s&method=getinfo" % (
self.base_url,
api,
SynoDSMInformation.API_KEY,
version,
)
self._information.update(self._get_url(url))
if self._utilisation:
api = "SYNO.Core.System.Utilization"
url = "%s/entry.cgi?api=%s&version=1&method=get&_sid=%s" % (
self.base_url,
api,
SynoCoreUtilization.API_KEY,
self.access_token,
)
self._utilisation.update(self._get_url(url))
if self._storage:
if self._dsm_version != 5:
api = "SYNO.Storage.CGI.Storage"
url = "%s/entry.cgi?api=%s&version=1&method=load_info&_sid=%s" % (
self.base_url,
api,
SynoStorage.API_KEY,
self.access_token,
)
self._storage.update(self._get_url(url))
else:
url = "%s?action=load_info&_sid=%s" % (
self.storage_url,
self._storage_url,
self.access_token,
)
output = self._get_url(url)["data"]
@ -209,13 +206,12 @@ class SynologyDSM(object):
def information(self):
"""Getter for various Information variables."""
if self._information is None:
api = "SYNO.DSM.Info"
version = 1
if self._dsm_version >= 6:
version = 2
url = "%s/entry.cgi?api=%s&version=%s&method=getinfo" % (
self.base_url,
api,
SynoDSMInformation.API_KEY,
version,
)
self._information = SynoDSMInformation(self._get_url(url))
@ -225,8 +221,10 @@ class SynologyDSM(object):
def utilisation(self):
"""Getter for various Utilisation variables."""
if self._utilisation is None:
api = "SYNO.Core.System.Utilization"
url = "%s/entry.cgi?api=%s&version=1&method=get" % (self.base_url, api)
url = "%s/entry.cgi?api=%s&version=1&method=get" % (
self.base_url,
SynoCoreUtilization.API_KEY,
)
self._utilisation = SynoCoreUtilization(self._get_url(url))
return self._utilisation
@ -235,13 +233,12 @@ class SynologyDSM(object):
"""Getter for various Storage variables."""
if self._storage is None:
if self._dsm_version != 5:
api = "SYNO.Storage.CGI.Storage"
url = "%s/entry.cgi?api=%s&version=1&method=load_info" % (
self.base_url,
api,
SynoStorage.API_KEY,
)
else:
url = "%s?action=load_info" % self.storage_url
url = "%s?action=load_info" % self._storage_url
output = self._get_url(url)
if self._dsm_version == 5:

63
tests/__init__.py Normal file
View File

@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
"""Library tests."""
from synology_dsm import SynologyDSM
from synology_dsm.api.core.utilization import SynoCoreUtilization
from synology_dsm.api.dsm.information import SynoDSMInformation
from synology_dsm.api.storage.storage import SynoStorage
from .const import DSM_6_LOGIN, DSM_6_INFORMATION, DSM_6_UTILIZATION, DSM_6_STORAGE
VALID_DSM_HOST = "nas.mywebsite.me"
VALID_DSM_PORT = "443"
VALID_USER = "valid_user"
VALID_PASSWORD = "valid_password"
class SynologyDSMMock(SynologyDSM):
"""Mocked SynologyDSM."""
LOGIN_URI = "auth.cgi"
API_URI = "entry.cgi"
def __init__(
self,
dsm_ip,
dsm_port,
username,
password,
use_https=False,
debugmode=False,
dsm_version=6,
):
SynologyDSM.__init__(
self,
dsm_ip,
dsm_port,
username,
password,
use_https,
debugmode,
dsm_version,
)
def _execute_get_url(self, request_url, append_sid=True):
if VALID_DSM_HOST not in request_url or VALID_DSM_PORT not in request_url:
return None
if (
self.LOGIN_URI in request_url
and VALID_USER in request_url
and VALID_PASSWORD in request_url
):
return DSM_6_LOGIN
if self.API_URI in request_url:
if SynoDSMInformation.API_KEY in request_url:
return DSM_6_INFORMATION
if SynoCoreUtilization.API_KEY in request_url:
return DSM_6_UTILIZATION
if SynoStorage.API_KEY in request_url:
return DSM_6_STORAGE
return None

473
tests/const.py Normal file
View File

@ -0,0 +1,473 @@
# -*- coding: utf-8 -*-
"""Test constants."""
# Name constants like this :
# "DSM_[dsm_version]_[API_KEY]"
# if data failed, add "_FAILED"
SID = "access_token"
SERIAL = "1x2X3x!_SN"
UNIQUE_KEY = "1x2X3x!_UK"
# DSM 6 RAW DATA
DSM_6_LOGIN = {"data": {"sid": SID}, "success": True}
DSM_6_INFORMATION = {
"data": {
"codepage": "fre",
"model": "DS918+",
"ram": 4096,
"serial": SERIAL,
"temperature": 40,
"temperature_warn": False,
"time": "Sun Mar 29 19:33:41 2020",
"uptime": 155084,
"version": "24922",
"version_string": "DSM 6.2.2-24922 Update 4",
},
"success": True,
}
DSM_6_UTILIZATION = {
"data": {
"cpu": {
"15min_load": 51,
"1min_load": 37,
"5min_load": 33,
"device": "System",
"other_load": 3,
"system_load": 0,
"user_load": 4,
},
"disk": {
"disk": [
{
"device": "sdc",
"display_name": "Drive 3",
"read_access": 3,
"read_byte": 55261,
"type": "internal",
"utilization": 12,
"write_access": 15,
"write_byte": 419425,
},
{
"device": "sda",
"display_name": "Drive 1",
"read_access": 3,
"read_byte": 63905,
"type": "internal",
"utilization": 8,
"write_access": 14,
"write_byte": 414795,
},
{
"device": "sdb",
"display_name": "Drive 2",
"read_access": 3,
"read_byte": 55891,
"type": "internal",
"utilization": 10,
"write_access": 15,
"write_byte": 415658,
},
],
"total": {
"device": "total",
"read_access": 9,
"read_byte": 175057,
"utilization": 10,
"write_access": 44,
"write_byte": 1249878,
},
},
"lun": [],
"memory": {
"avail_real": 156188,
"avail_swap": 4146316,
"buffer": 15172,
"cached": 2764756,
"device": "Memory",
"memory_size": 4194304,
"real_usage": 24,
"si_disk": 0,
"so_disk": 0,
"swap_usage": 6,
"total_real": 3867268,
"total_swap": 4415404,
},
"network": [
{"device": "total", "rx": 109549, "tx": 45097},
{"device": "eth0", "rx": 109549, "tx": 45097},
{"device": "eth1", "rx": 0, "tx": 0},
],
"space": {
"total": {
"device": "total",
"read_access": 1,
"read_byte": 27603,
"utilization": 1,
"write_access": 23,
"write_byte": 132496,
},
"volume": [
{
"device": "md2",
"display_name": "volume1",
"read_access": 1,
"read_byte": 27603,
"utilization": 1,
"write_access": 23,
"write_byte": 132496,
}
],
},
"time": 1585503221,
},
"success": True,
}
DSM_6_STORAGE = {
"data": {
"disks": [
{"id": "test_disk"},
{
"adv_progress": "",
"adv_status": "normal",
"below_remain_life_thr": False,
"compatibility": "disabled",
"container": {
"order": 0,
"str": "DS918+",
"supportPwrBtnDisable": False,
"type": "internal",
},
"device": "/dev/sda",
"disable_secera": False,
"diskType": "SATA",
"disk_code": "ironwolf",
"erase_time": 448,
"exceed_bad_sector_thr": False,
"firm": "SC60",
"has_system": True,
"id": "sda",
"ihm_testing": False,
"is4Kn": False,
"isSsd": False,
"isSynoPartition": True,
"is_erasing": False,
"longName": "Drive 1",
"model": "ST4000VN008-2DR166 ",
"name": "Drive 1",
"num_id": 1,
"order": 1,
"overview_status": "normal",
"pciSlot": -1,
"perf_testing": False,
"portType": "normal",
"remain_life": -1,
"remote_info": {"compatibility": "disabled", "unc": 0},
"serial": "ZDH4LYTS",
"size_total": "4000787030016",
"smart_progress": "",
"smart_status": "normal",
"smart_test_limit": 0,
"smart_testing": False,
"status": "normal",
"support": False,
"temp": 24,
"testing_progress": "",
"testing_type": "",
"tray_status": "join",
"unc": 0,
"used_by": "reuse_1",
"vendor": "Seagate",
},
{
"adv_progress": "",
"adv_status": "normal",
"below_remain_life_thr": False,
"compatibility": "disabled",
"container": {
"order": 0,
"str": "DS918+",
"supportPwrBtnDisable": False,
"type": "internal",
},
"device": "/dev/sdb",
"disable_secera": False,
"diskType": "SATA",
"disk_code": "ironwolf",
"erase_time": 448,
"exceed_bad_sector_thr": False,
"firm": "SC60",
"has_system": True,
"id": "sdb",
"ihm_testing": False,
"is4Kn": False,
"isSsd": False,
"isSynoPartition": True,
"is_erasing": False,
"longName": "Drive 2",
"model": "ST4000VN008-2DR166 ",
"name": "Drive 2",
"num_id": 2,
"order": 2,
"overview_status": "normal",
"pciSlot": -1,
"perf_testing": False,
"portType": "normal",
"remain_life": -1,
"remote_info": {"compatibility": "disabled", "unc": 0},
"serial": "ZDH4LS72",
"size_total": "4000787030016",
"smart_progress": "",
"smart_status": "normal",
"smart_test_limit": 0,
"smart_testing": False,
"status": "normal",
"support": False,
"temp": 24,
"testing_progress": "",
"testing_type": "",
"tray_status": "join",
"unc": 0,
"used_by": "reuse_1",
"vendor": "Seagate",
},
{
"adv_progress": "",
"adv_status": "normal",
"below_remain_life_thr": False,
"compatibility": "disabled",
"container": {
"order": 0,
"str": "DS918+",
"supportPwrBtnDisable": False,
"type": "internal",
},
"device": "/dev/sdc",
"disable_secera": False,
"diskType": "SATA",
"disk_code": "ironwolf",
"erase_time": 452,
"exceed_bad_sector_thr": False,
"firm": "SC60",
"has_system": True,
"id": "sdc",
"ihm_testing": False,
"is4Kn": False,
"isSsd": False,
"isSynoPartition": True,
"is_erasing": False,
"longName": "Drive 3",
"model": "ST4000VN008-2DR166 ",
"name": "Drive 3",
"num_id": 3,
"order": 3,
"overview_status": "normal",
"pciSlot": -1,
"perf_testing": False,
"portType": "normal",
"remain_life": -1,
"remote_info": {"compatibility": "disabled", "unc": 0},
"serial": "ZDH4LQ1H",
"size_total": "4000787030016",
"smart_progress": "",
"smart_status": "normal",
"smart_test_limit": 0,
"smart_testing": False,
"status": "normal",
"support": False,
"temp": 23,
"testing_progress": "",
"testing_type": "",
"tray_status": "join",
"unc": 0,
"used_by": "reuse_1",
"vendor": "Seagate",
},
],
"env": {
"batchtask": {"max_task": 64, "remain_task": 64},
"bay_number": "4",
"data_scrubbing": {"sche_enabled": "0", "sche_status": "disabled",},
"ebox": [],
"fs_acting": False,
"isSyncSysPartition": False,
"is_space_actioning": False,
"isns": {"address": "", "enabled": False},
"isns_server": "",
"max_fs_bytes": "118747255799808",
"max_fs_bytes_high_end": "219902325555200",
"model_name": "DS918+",
"ram_enough_for_fs_high_end": False,
"ram_size": 4,
"ram_size_required": 32,
"showpooltab": False,
"status": {"system_crashed": False, "system_need_repair": False,},
"support": {"ebox": True, "raid_cross": True, "sysdef": True,},
"support_fit_fs_limit": True,
"unique_key": UNIQUE_KEY,
"volume_full_critical": 0.1,
"volume_full_warning": 0.2,
},
"hotSpareConf": {"cross_repair": True, "disable_repair": []},
"hotSpares": [],
"iscsiLuns": [],
"iscsiTargets": [],
"ports": [],
"ssdCaches": [],
"storagePools": [
{
"cacheStatus": "",
"can_do": {
"data_scrubbing": True,
"delete": True,
"expand_by_disk": 1,
"migrate": {"to_raid5+spare": "1-1", "to_raid6": 1},
"raid_cross": True,
},
"container": "internal",
"deploy_path": "volume_1",
"desc": "Situé sur Groupe de stockage 1, RAID 5",
"device_type": "raid_5",
"disk_failure_number": 0,
"disks": ["sda", "sdb", "sdc"],
"drive_type": 0,
"id": "reuse_1",
"is_actioning": False,
"is_scheduled": False,
"is_writable": True,
"last_done_time": 1551201018,
"limited_disk_number": 24,
"maximal_disk_size": "0",
"minimal_disk_size": "4000681164800",
"next_schedule_time": 0,
"num_id": 1,
"pool_path": "reuse_1",
"progress": {"percent": "-1", "step": "none"},
"raidType": "single",
"raids": [
{
"designedDiskCount": 3,
"devices": [
{"id": "sdc", "slot": 2, "status": "normal",},
{"id": "sdb", "slot": 1, "status": "normal",},
{"id": "sda", "slot": 0, "status": "normal",},
],
"hasParity": True,
"minDevSize": "4000681164800",
"normalDevCount": 3,
"raidPath": "/dev/md2",
"raidStatus": 1,
"spares": [],
}
],
"scrubbingStatus": "no_action",
"size": {"total": "7991698522112", "used": "7991698522112",},
"space_path": "/dev/md2",
"ssd_trim": {"support": "not support"},
"status": "normal",
"suggestions": [],
"timebackup": False,
"vspace_can_do": {
"drbd": {
"resize": {
"can_do": False,
"errCode": 53504,
"stopService": False,
}
},
"flashcache": {
"apply": {"can_do": True, "errCode": 0, "stopService": True,},
"remove": {"can_do": True, "errCode": 0, "stopService": False,},
"resize": {"can_do": True, "errCode": 0, "stopService": False,},
},
"snapshot": {
"resize": {
"can_do": False,
"errCode": 53504,
"stopService": False,
}
},
},
}
],
"volumes": [
{"id": "test_volume"},
{
"atime_checked": False,
"atime_opt": "relatime",
"cacheStatus": "",
"can_do": {
"data_scrubbing": True,
"delete": True,
"expand_by_disk": 1,
"migrate": {"to_raid5+spare": "1-1", "to_raid6": 1},
"raid_cross": True,
},
"container": "internal",
"deploy_path": "volume_1",
"desc": "Situé sur Groupe de stockage 1, RAID 5",
"device_type": "raid_5",
"disk_failure_number": 0,
"disks": [],
"drive_type": 0,
"eppool_used": "0",
"exist_alive_vdsm": False,
"fs_type": "btrfs",
"id": "volume_1",
"is_acting": False,
"is_actioning": False,
"is_inode_full": False,
"is_scheduled": False,
"is_writable": True,
"last_done_time": 1551201018,
"limited_disk_number": 24,
"max_fs_size": "1152921504606846976",
"next_schedule_time": 0,
"num_id": 1,
"pool_path": "reuse_1",
"progress": {"percent": "-1", "step": "none"},
"raidType": "single",
"scrubbingStatus": "no_action",
"size": {
"free_inode": "0",
"total": "7672030584832",
"total_device": "7991698522112",
"total_inode": "0",
"used": "4377452806144",
},
"ssd_trim": {"support": "not support"},
"status": "normal",
"suggestions": [],
"timebackup": False,
"used_by_gluster": False,
"vol_path": "/volume1",
"vspace_can_do": {
"drbd": {
"resize": {
"can_do": False,
"errCode": 53504,
"stopService": False,
}
},
"flashcache": {
"apply": {"can_do": True, "errCode": 0, "stopService": True,},
"remove": {"can_do": True, "errCode": 0, "stopService": False,},
"resize": {"can_do": True, "errCode": 0, "stopService": False,},
},
"snapshot": {
"resize": {
"can_do": False,
"errCode": 53504,
"stopService": False,
}
},
},
},
],
},
"success": True,
}

186
tests/test_synology_dsm.py Normal file
View File

@ -0,0 +1,186 @@
# -*- coding: utf-8 -*-
"""Synology DSM tests."""
from unittest import TestCase
from . import (
SynologyDSMMock,
VALID_DSM_HOST,
VALID_DSM_PORT,
VALID_PASSWORD,
VALID_USER,
)
from .const import SID, SERIAL
class TestSynologyDSM(TestCase):
"""SynologyDSM test cases."""
api = None
def setUp(self):
self.api = SynologyDSMMock(
VALID_DSM_HOST, VALID_DSM_PORT, VALID_USER, VALID_PASSWORD
)
def test_init(self):
"""Test init."""
assert self.api.username
assert not self.api.access_token
assert self.api.base_url
def test_login(self):
"""Test login."""
assert self.api.login()
assert self.api.access_token == SID
def test_login_failed(self): # pylint: disable=no-self-use
"""Test failed login."""
api = SynologyDSMMock("host", VALID_DSM_PORT, VALID_USER, VALID_PASSWORD)
assert not api.login()
assert not api.access_token
api = SynologyDSMMock(VALID_DSM_HOST, 0, VALID_USER, VALID_PASSWORD)
assert not api.login()
assert not api.access_token
api = SynologyDSMMock(VALID_DSM_HOST, VALID_DSM_PORT, "user", VALID_PASSWORD)
assert not api.login()
assert not api.access_token
api = SynologyDSMMock(VALID_DSM_HOST, VALID_DSM_PORT, VALID_USER, "pass")
assert not api.login()
assert not api.access_token
def test_information(self):
"""Test information."""
assert self.api.information
assert self.api.information.model == "DS918+"
assert self.api.information.ram == 4096
assert self.api.information.serial == SERIAL
assert self.api.information.temperature == 40
assert not self.api.information.temperature_warn
assert self.api.information.uptime == 155084
assert self.api.information.version_string == "DSM 6.2.2-24922 Update 4"
def test_utilisation(self):
"""Test utilization."""
assert self.api.utilisation
def test_utilisation_cpu(self):
"""Test utilization CPU."""
assert self.api.utilisation.cpu
assert self.api.utilisation.cpu_other_load
assert self.api.utilisation.cpu_user_load
assert self.api.utilisation.cpu_system_load == 0
assert self.api.utilisation.cpu_total_load
assert self.api.utilisation.cpu_1min_load
assert self.api.utilisation.cpu_5min_load
assert self.api.utilisation.cpu_15min_load
def test_utilisation_memory(self):
"""Test utilization memory."""
assert self.api.utilisation.memory
assert self.api.utilisation.memory_real_usage
assert self.api.utilisation.memory_size
assert self.api.utilisation.memory_available_swap
assert self.api.utilisation.memory_cached
assert self.api.utilisation.memory_available_real
assert self.api.utilisation.memory_total_real
assert self.api.utilisation.memory_total_swap
def test_utilisation_network(self):
"""Test utilization network."""
assert self.api.utilisation.network
assert self.api.utilisation.network_up
assert self.api.utilisation.network_down
def test_storage(self):
"""Test storage roots."""
assert self.api.storage
assert self.api.storage.disks
assert self.api.storage.env
assert self.api.storage.storage_pools
assert self.api.storage.volumes
def test_storage_volumes(self):
"""Test storage volumes."""
# Basics
assert self.api.storage.volumes_ids
for volume_id in self.api.storage.volumes_ids:
if volume_id == "test_volume":
continue
assert self.api.storage.volume_status(volume_id)
assert self.api.storage.volume_device_type(volume_id)
assert self.api.storage.volume_size_total(volume_id)
assert self.api.storage.volume_size_total(volume_id, False)
assert self.api.storage.volume_size_used(volume_id)
assert self.api.storage.volume_size_used(volume_id, False)
assert self.api.storage.volume_percentage_used(volume_id)
assert self.api.storage.volume_disk_temp_avg(volume_id)
assert self.api.storage.volume_disk_temp_max(volume_id)
# Existing volume
assert self.api.storage.volume_status("volume_1") == "normal"
assert self.api.storage.volume_device_type("volume_1") == "raid_5"
assert self.api.storage.volume_size_total("volume_1") == "7.0Tb"
assert self.api.storage.volume_size_total("volume_1", False) == 7672030584832
assert self.api.storage.volume_size_used("volume_1") == "4.0Tb"
assert self.api.storage.volume_size_used("volume_1", False) == 4377452806144
assert self.api.storage.volume_percentage_used("volume_1") == 57.1
assert self.api.storage.volume_disk_temp_avg("volume_1") == 24.0
assert self.api.storage.volume_disk_temp_max("volume_1") == 24
# Non existing volume
assert not self.api.storage.volume_status("not_a_volume")
assert not self.api.storage.volume_device_type("not_a_volume")
assert not self.api.storage.volume_size_total("not_a_volume")
assert not self.api.storage.volume_size_total("not_a_volume", False)
assert not self.api.storage.volume_size_used("not_a_volume")
assert not self.api.storage.volume_size_used("not_a_volume", False)
assert not self.api.storage.volume_percentage_used("not_a_volume")
assert not self.api.storage.volume_disk_temp_avg("not_a_volume")
assert not self.api.storage.volume_disk_temp_max("not_a_volume")
# Test volume
assert self.api.storage.volume_status("test_volume") is None
assert self.api.storage.volume_device_type("test_volume") is None
assert self.api.storage.volume_size_total("test_volume") is None
assert self.api.storage.volume_size_total("test_volume", False) is None
assert self.api.storage.volume_size_used("test_volume") is None
assert self.api.storage.volume_size_used("test_volume", False) is None
assert self.api.storage.volume_percentage_used("test_volume") is None
assert self.api.storage.volume_disk_temp_avg("test_volume") is None
assert self.api.storage.volume_disk_temp_max("test_volume") is None
def test_storage_disks(self):
"""Test storage disks."""
# Basics
assert self.api.storage.disks_ids
for disk_id in self.api.storage.disks_ids:
if disk_id == "test_disk":
continue
assert "Drive" in self.api.storage.disk_name(disk_id)
assert "/dev/" in self.api.storage.disk_device(disk_id)
assert self.api.storage.disk_smart_status(disk_id) == "normal"
assert self.api.storage.disk_status(disk_id) == "normal"
assert not self.api.storage.disk_exceed_bad_sector_thr(disk_id)
assert not self.api.storage.disk_below_remain_life_thr(disk_id)
assert self.api.storage.disk_temp(disk_id)
# Non existing disk
assert not self.api.storage.disk_name("not_a_disk")
assert not self.api.storage.disk_device("not_a_disk")
assert not self.api.storage.disk_smart_status("not_a_disk")
assert not self.api.storage.disk_status("not_a_disk")
assert not self.api.storage.disk_exceed_bad_sector_thr("not_a_disk")
assert not self.api.storage.disk_below_remain_life_thr("not_a_disk")
assert not self.api.storage.disk_temp("not_a_disk")
# Test disk
assert self.api.storage.disk_name("test_disk") is None
assert self.api.storage.disk_device("test_disk") is None
assert self.api.storage.disk_smart_status("test_disk") is None
assert self.api.storage.disk_status("test_disk") is None
assert self.api.storage.disk_exceed_bad_sector_thr("test_disk") is None
assert self.api.storage.disk_below_remain_life_thr("test_disk") is None
assert self.api.storage.disk_temp("test_disk") is None