Commit df39a86c authored by Jondy Zhao's avatar Jondy Zhao

Add testcases for netreport module

parent d43a8995
...@@ -32,6 +32,11 @@ Then run test: ...@@ -32,6 +32,11 @@ Then run test:
$ cd src $ cd src
$ make test $ make test
Before test netreport.py,
$ easy_install lxml
$ ln -s /opt/git/slapos.core
Use Cases Use Cases
========= =========
......
...@@ -18,7 +18,6 @@ def get_description(): ...@@ -18,7 +18,6 @@ def get_description():
VERSION = "0.1.1" VERSION = "0.1.1"
if sys.platform.startswith("cygwin"): if sys.platform.startswith("cygwin"):
def get_winver(): def get_winver():
...@@ -42,10 +41,10 @@ else: ...@@ -42,10 +41,10 @@ else:
def main(): def main():
setup_args = dict( setup_args = dict(
name='netreport', name='netdrive.report',
version=VERSION, version=VERSION,
download_url='http://', download_url='http://',
description='A tool used to report the usage of net resource in the Windows', description='A tool used to report the usage of net drive in the Windows',
long_description=get_description(), long_description=get_description(),
keywords=['netdrive',], keywords=['netdrive',],
scripts=['src/netreport.py'], scripts=['src/netreport.py'],
...@@ -54,8 +53,17 @@ def main(): ...@@ -54,8 +53,17 @@ def main():
maintainer='Jondy Zhao', maintainer='Jondy Zhao',
maintainer_email='jondy.zhao@nexedi.com', maintainer_email='jondy.zhao@nexedi.com',
url='http://', url='http://',
platforms='Platform Independent', license='GPLv3',
license='License :: OSI Approved :: BSD License', zip_safe=False,
install_requires=[
'lxml',
'slapos.core',
],
entry_points = {
'console_scripts': [
'netdrive-reporter = netreport:main',
],
}
) )
if extensions is not None: if extensions is not None:
setup_args["ext_modules"] = extensions setup_args["ext_modules"] = extensions
......
...@@ -4,57 +4,141 @@ import sys ...@@ -4,57 +4,141 @@ import sys
import os import os
import tempfile import tempfile
import shutil import shutil
import datetime
sys.path.insert(0, 'slapos.core')
sys.path.insert(0, os.getcwd()) sys.path.insert(0, os.getcwd())
import test.test_support import test.test_support
netuse = test.test_support.import_module('netuse') real_netuse = test.test_support.import_module('netuse')
netreport = test.test_support.import_module('netreport')
threading = test.test_support.import_module('threading') threading = test.test_support.import_module('threading')
import unittest import unittest
class BaseTestCase(unittest.TestCase): class NetUseTests(unittest.TestCase):
def setUp(self):
self._threads = test.test_support.threading_setup()
def tearDown(self):
test.test_support.threading_cleanup(*self._threads)
test.test_support.reap_children()
class NetUsageTests(BaseTestCase):
def test_user_info(self): def test_user_info(self):
u = netuse.userInfo() r = real_netuse.userInfo()
self.assertEquals(len(u), 3) self.assertEquals(len(r), 3)
self.assertEquals(u, []) self.assertEquals(r, ('JONDY', 'JONDY', 'Administrator'))
def test_usage_report(self): def test_usage_report(self):
r = netuse.usageReport() r = real_netuse.usageReport()
self.assertEquals(len(r), 0) self.assertEquals(len(r), 0)
self.assertEquals(r, []) self.assertEquals(r, [])
def test_usage_report_server(self): def test_usage_report_server(self):
r = netuse.usageReport('myserver') r = real_netuse.usageReport('myserver')
self.assertEquals(len(r), 0) self.assertEquals(len(r), 0)
self.assertEquals(r, []) self.assertEquals(r, [])
def test_usage_report_server_is_none(self): def test_usage_report_server_is_none(self):
r = netuse.usageReport(None) r = real_netuse.usageReport()
self.assertEquals(len(r), 0) self.assertEquals(len(r), 0)
self.assertEquals(r, []) self.assertEquals(r, [])
def test_usage_report_one_share_folder_to_many_drive(self): def test_usage_report_one_share_folder_to_many_drive(self):
r = netuse.usageReport(None) r = real_netuse.usageReport()
self.assertEquals(len(r), 0) self.assertEquals(len(r), 0)
self.assertEquals(r, []) self.assertEquals(r, [])
def test_usage_report_share_subfolder(self): def test_usage_report_share_subfolder(self):
r = netuse.usageReport(None) r = real_netuse.usageReport()
self.assertEquals(len(r), 0) self.assertEquals(len(r), 0)
self.assertEquals(r, []) self.assertEquals(r, [])
class BaseTestCase(unittest.TestCase):
def setUp(self):
self._threads = test.test_support.threading_setup()
self._db_file = '/tmp/mytest.db'
options = { 'master_url' : 'http://localhost:12006',
'computer_id' : 'COMP-1500',
'server_name' : '',
'report_interval' : 300,
'data_file' : self._db_file,
}
self.reporter = netreport.NetDriveUsageReporter(options)
self.db = self.reporter._db
def tearDown(self):
test.test_support.threading_cleanup(*self._threads)
test.test_support.reap_children()
self.db.close()
if os.path.exists(self._db_file):
os.remove(self._db_file)
class NetReportTests(BaseTestCase):
def test_parse_argument(self):
argv_orig = sys.argv[:]
sys.argv = ['me',
'--master-url', 'http://127.0.0.1',
'--computer-id', 'COMP-1500',
'--cert-file', '~/certificate',
'--key-file', '~/key',
'--report-interval', '500',
'--data-file', '/tmp/myreporter',
]
opts = netreport.parseArgumentTuple()
self.assertEquals(opts,
{'cert_file': '~/certificate',
'computer_id': 'COMP-1500',
'data_file': '/tmp/myreporter',
'key_file': '~/key',
'master_url': 'http://127.0.0.1',
'report_interval': '500',
'server_name': ''}
)
sys.argv[:] = argv_orig
def test_init_config(self):
self.reporter.initializeConfigData()
self.assertEquals(self.reporter._config_id, 1)
self.assertEquals(self.reporter._domain_account, 'JONDY\\Administrator')
self.assertEquals(self.reporter._report_date, datetime.date.today().isoformat())
def test_insert_report(self):
self.reporter.initializeConfigData()
t = datetime.datetime.now().isoformat()
self.reporter.insertUsageReport(t, 500)
def test_send_all_report(self):
self.reporter.initializeConfigData()
self.reporter.sendAllReport()
def test_generate_daily_report_no_data(self):
self.reporter.initializeConfigData()
self.reporter.insertUsageReport('2011-01-02 12:23:32', 500)
xml = self.reporter.generateDailyReport(1,
'COMP-1500',
'SERVER\\JONDY',
'2021-03-05',
)
self.assertEquals(xml, None)
def test_generate_daily_report_no_data(self):
self.reporter.initializeConfigData()
self.reporter.insertUsageReport('2011-01-02 12:23:32', 500)
report_date = '2021-03-05'
config_id = self.reporter._config_id
q = self.db.execute
q( "INSERT INTO net_drive_usage "
"(config_id, drive_letter, remote_folder, "
" start, duration, usage_bytes )"
" VALUES (?, ?, ?, ?, ?, ?)",
(config_id, 'E:', r'\Server\Path', '2021-03-05 12:23:30', 300, 1000))
xml = self.reporter.generateDailyReport(config_id,
self.reporter.computer_id,
'JONDY\\Adminstrator',
report_date,
)
self.assertEquals(xml, "<?xml version='1.0' encoding='utf-8'?>\n<report><computer>COMP-1500</computer><account>JONDY\\Adminstrator</account><date>2021-03-05</date><usage>300000.0</usage><details/></report>")
if __name__ == "__main__": if __name__ == "__main__":
# unittest.main() # unittest.main()
loader = unittest.TestLoader() loader = unittest.TestLoader()
# loader.testMethodPrefix = 'test_' # loader.testMethodPrefix = 'test_'
suite = loader.loadTestsFromTestCase(NetUsageTests) suite1 = loader.loadTestsFromTestCase(NetUseTests)
unittest.TextTestRunner(verbosity=2).run(suite) suite2 = loader.loadTestsFromTestCase(NetReportTests)
alltests = unittest.TestSuite([suite1, suite2])
unittest.TextTestRunner(verbosity=2).run(alltests)
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment