code:
import collections
import lib.Hydrus.Constants as HC
import os
import random
import time
import unittest
import lib.Hydrus.Data
import lib.Hydrus.Globals as HG
import lib.Networking
from mock import patch

now = lib.Hydrus.Data.GetNow()

now_10 = now + 10

now_20 = now + 20

with patch.object( HydrusData, 'GetNow', return_value = now ):
	HIGH_USAGE = lib.Networking.BandwidthTracker()

	for i in range( 100 ):
		HIGH_USAGE.ReportRequestUsed()
		HIGH_USAGE.ReportDataUsed( random.randint( 512, 1024 ) )

	LOW_USAGE = lib.Networking.BandwidthTracker()

	LOW_USAGE.ReportRequestUsed()
	LOW_USAGE.ReportDataUsed( 1024 )

	ZERO_USAGE = lib.Networking.BandwidthTracker()

class TestBandwidthRules( unittest.TestCase ):
	def test_no_rules( self ):
		rules = lib.Networking.BandwidthRules()

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

	def test_per_sec( self ):
		# at short time deltas, we can always start based on data alone

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_DATA, 1, 10240 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertFalse( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		#

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, 1, 1 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertFalse( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		#

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_DATA, 1, 10240 )
		rules.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, 1, 1 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertFalse( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertFalse( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertTrue( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

	def test_per_min( self ):
		# cutoff is 15s for continue

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_DATA, 60, 10240 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		#

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, 60, 10 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		#

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_DATA, 60, 10240 )
		rules.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, 60, 10 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

	def test_per_month( self ):
		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 10240 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		#

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, None, 10 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		#

		rules = lib.Networking.BandwidthRules()

		rules.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 10240 )
		rules.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, None, 10 )

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_10 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

		with patch.object( HydrusData, 'GetNow', return_value = now_20 ):
			self.assertTrue( rules.CanStartRequest( ZERO_USAGE ) )
			self.assertTrue( rules.CanStartRequest( LOW_USAGE ) )
			self.assertFalse( rules.CanStartRequest( HIGH_USAGE ) )

			self.assertTrue( rules.CanContinueDownload( ZERO_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( LOW_USAGE ) )
			self.assertTrue( rules.CanContinueDownload( HIGH_USAGE ) )

class TestBandwidthTracker( unittest.TestCase ):
	def test_bandwidth_tracker( self ):
		bandwidth_tracker = lib.Networking.BandwidthTracker()

		self.assertEqual( bandwidth_tracker.GetCurrentMonthSummary(), 'used 0B in 0 requests this month' )

		now = lib.Hydrus.Data.GetNow()

		with patch.object( HydrusData, 'GetNow', return_value = now ):
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 0 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 0 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 1 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 1 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 2 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 2 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 6 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 6 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 3600 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 3600 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, None ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, None ), 0 )

			#

			bandwidth_tracker.ReportDataUsed( 1024 )
			bandwidth_tracker.ReportRequestUsed()

			self.assertEqual( bandwidth_tracker.GetCurrentMonthSummary(), 'used 1.0KB in 1 requests this month' )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 0 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 0 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 1 ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 1 ), 1 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 2 ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 2 ), 1 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 6 ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 6 ), 1 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 3600 ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 3600 ), 1 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, None ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, None ), 1 )

		#

		five_secs_from_now = now + 5

		with patch.object( HydrusData, 'GetNow', return_value = five_secs_from_now ):
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 0 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 0 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 1 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 1 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 2 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 2 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 6 ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 6 ), 1 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 3600 ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 3600 ), 1 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, None ), 1024 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, None ), 1 )

			#

			bandwidth_tracker.ReportDataUsed( 32 )
			bandwidth_tracker.ReportRequestUsed()

			bandwidth_tracker.ReportDataUsed( 32 )
			bandwidth_tracker.ReportRequestUsed()

			self.assertEqual( bandwidth_tracker.GetCurrentMonthSummary(), 'used 1.1KB in 3 requests this month' )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 0 ), 0 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 0 ), 0 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 1 ), 64 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 1 ), 2 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 2 ), 64 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 2 ), 2 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 6 ), 1088 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 6 ), 3 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, 3600 ), 1088 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, 3600 ), 3 )

			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, None ), 1088 )
			self.assertEqual( bandwidth_tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, None ), 3 )


code:
import lib.Hydrus.Caches
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.Data.Local
import lib.Hydrus.Defaults
import lib.Networking.Downloading
import lib.Hydrus.Duplicates
import lib.GUI.Shortcuts
import lib.Actions.Import
import lib.Actions.Import.Options
import lib.Actions.Import.Subscriptions
import lib.Hydrus.Media
import lib.Networking.Domain
import lib.Hydrus.Ratings
import lib.Hydrus.Search
import lib.Hydrus.Tags.Local
import lib.Hydrus.Constants as HC
import lib.Hydrus.Data
import lib.Networking
import lib.Hydrus.Serialisable
from . import TestController as TC
import os
import unittest
import wx

class TestSerialisables( unittest.TestCase ):
	def _dump_and_load_and_test( self, obj, test_func ):
		serialisable_tuple = obj.GetSerialisableTuple()

		self.assertIsInstance( serialisable_tuple, tuple )

		if isinstance( obj, lib.Hydrus.Serialisable.SerialisableBaseNamed ):
			( serialisable_type, name, version, serialisable_info ) = serialisable_tuple

		elif isinstance( obj, lib.Hydrus.Serialisable.SerialisableBase ):
			( serialisable_type, version, serialisable_info ) = serialisable_tuple

		self.assertEqual( serialisable_type, obj.SERIALISABLE_TYPE )
		self.assertEqual( version, obj.SERIALISABLE_VERSION )

		dupe_obj = lib.Hydrus.Serialisable.CreateFromSerialisableTuple( serialisable_tuple )

		self.assertIsNot( obj, dupe_obj )

		test_func( obj, dupe_obj )

		#

		json_string = obj.DumpToString()

		self.assertIsInstance( json_string, str )

		dupe_obj = lib.Hydrus.Serialisable.CreateFromString( json_string )

		self.assertIsNot( obj, dupe_obj )

		test_func( obj, dupe_obj )

		#

		network_bytes = obj.DumpToNetworkBytes()

		self.assertIsInstance( network_bytes, bytes )

		dupe_obj = lib.Hydrus.Serialisable.CreateFromNetworkBytes( network_bytes )

		self.assertIsNot( obj, dupe_obj )

		test_func( obj, dupe_obj )

	def test_basics( self ):
		def test( obj, dupe_obj ):
			self.assertEqual( len( list( obj.items() ) ), len( list( dupe_obj.items() ) ) )

			for ( key, value ) in list( obj.items() ):
				self.assertEqual( value, dupe_obj[ key ] )

		#

		d = lib.Hydrus.Serialisable.SerialisableDictionary()

		d[ 1 ] = 2
		d[ 3 ] = 'test1'

		d[ 'test2' ] = 4
		d[ 'test3' ] = 5

		d[ 6 ] = lib.Hydrus.Serialisable.SerialisableDictionary( { i : 'test' + str( i ) for i in range( 20 ) } )
		d[ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'test pred 1' ) ] = 56

		d[ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'test pred 2' ) ] = lib.Hydrus.Serialisable.SerialisableList( [ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'test' + str( i ) ) for i in range( 10 ) ] )

		self.assertEqual( len( list( d.keys() ) ), 7 )

		for ( key, value ) in list( d.items() ):
			self.assertEqual( d[ key ], value )

		self._dump_and_load_and_test( d, test )

		#

		db = lib.Hydrus.Serialisable.SerialisableBytesDictionary()

		db[ lib.Hydrus.Data.GenerateKey() ] = lib.Hydrus.Data.GenerateKey()
		db[ lib.Hydrus.Data.GenerateKey() ] = [ lib.Hydrus.Data.GenerateKey() for i in range( 10 ) ]
		db[ 1 ] = lib.Hydrus.Data.GenerateKey()
		db[ 2 ] = [ lib.Hydrus.Data.GenerateKey() for i in range( 10 ) ]

		self.assertEqual( len( list( db.keys() ) ), 4 )

		for ( key, value ) in list( db.items() ):
			self.assertEqual( db[ key ], value )

		self._dump_and_load_and_test( db, test )

	def test_HG.SerialisableType["APPLICATION_COMMAND("] self ):
		def test( obj, dupe_obj ):
			self.assertEqual( obj.GetCommandType(), dupe_obj.GetCommandType() )

			self.assertEqual( obj.GetData(), dupe_obj.GetData() )

		acs = []

		acs.append( ( lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_SIMPLE, 'archive_file' ), 'archive_file' ) )
		acs.append( ( lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_CONTENT, ( lib.Hydrus.Data.GenerateKey(), HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_FLIP, 'test' ) ), 'flip on/off mappings "test" for unknown service!' ) )
		acs.append( ( lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_CONTENT, ( CC.LOCAL_TAG_SERVICE_KEY, HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_FLIP, 'test' ) ), 'flip on/off mappings "test" for local tags' ) )
		acs.append( ( lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_CONTENT, ( lib.Hydrus.Data.GenerateKey(), HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_SET, 0.4 ) ), 'set ratings "0.4" for unknown service!' ) )

		for ( ac, s ) in acs:
			self._dump_and_load_and_test( ac, test )

			self.assertEqual( ac.ToString(), s )

	def test_HG.SerialisableType["DUPLICATE_ACTION_OPTIONS("] self ):
		def test( obj, dupe_obj ):
			self.assertEqual( obj.ToTuple(), dupe_obj.ToTuple() )

		duplicate_action_options_delete_and_move = lib.Hydrus.Duplicates.DuplicateActionOptions( [ ( CC.LOCAL_TAG_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_MOVE, lib.Hydrus.Tags.Local.TagFilter() ) ], [ ( TC.LOCAL_RATING_LIKE_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_MOVE ), ( TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_MOVE ) ], True )
		duplicate_action_options_copy = lib.Hydrus.Duplicates.DuplicateActionOptions( [ ( CC.LOCAL_TAG_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_COPY, lib.Hydrus.Tags.Local.TagFilter() ) ], [ ( TC.LOCAL_RATING_LIKE_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_COPY ), ( TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_COPY ) ], False )
		duplicate_action_options_merge = lib.Hydrus.Duplicates.DuplicateActionOptions( [ ( CC.LOCAL_TAG_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_TWO_WAY_MERGE, lib.Hydrus.Tags.Local.TagFilter() ) ], [ ( TC.LOCAL_RATING_LIKE_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_TWO_WAY_MERGE ), ( TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY, HC.CONTENT_MERGE_ACTION_TWO_WAY_MERGE ) ], False )

		inbox = True
		size = 40960
		mime = HC.IMAGE_JPEG
		width = 640
		height = 480
		duration = None
		num_frames = None
		num_words = None

		local_locations_manager = lib.Hydrus.Media.LocationsManager( { CC.LOCAL_FILE_SERVICE_KEY, CC.COMBINED_LOCAL_FILE_SERVICE_KEY }, set(), set(), set(), inbox )
		trash_locations_manager = lib.Hydrus.Media.LocationsManager( { CC.TRASH_SERVICE_KEY, CC.COMBINED_LOCAL_FILE_SERVICE_KEY }, set(), set(), set(), inbox )
		deleted_locations_manager = lib.Hydrus.Media.LocationsManager( set(), { CC.COMBINED_LOCAL_FILE_SERVICE_KEY }, set(), set(), inbox )

		# duplicate to generate proper dicts

		one_tags_manager = lib.Hydrus.Media.TagsManager( { CC.LOCAL_TAG_SERVICE_KEY : { HC.CONTENT_STATUS_CURRENT : { 'one' } } } ).Duplicate()
		two_tags_manager = lib.Hydrus.Media.TagsManager( { CC.LOCAL_TAG_SERVICE_KEY : { HC.CONTENT_STATUS_CURRENT : { 'two' } } } ).Duplicate()
		substantial_tags_manager = lib.Hydrus.Media.TagsManager( { CC.LOCAL_TAG_SERVICE_KEY : { HC.CONTENT_STATUS_CURRENT : { 'test tag', 'series:namespaced test tag' } } } ).Duplicate()
		empty_tags_manager = lib.Hydrus.Media.TagsManager( {} ).Duplicate()

		one_ratings_manager = lib.Hydrus.Ratings.RatingsManager( { TC.LOCAL_RATING_LIKE_SERVICE_KEY : 1.0, TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY : 0.8 } )
		two_ratings_manager = lib.Hydrus.Ratings.RatingsManager( { TC.LOCAL_RATING_LIKE_SERVICE_KEY : 0.0, TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY : 0.6 } )
		substantial_ratings_manager = lib.Hydrus.Ratings.RatingsManager( { TC.LOCAL_RATING_LIKE_SERVICE_KEY : 1.0, TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY : 0.8 } )
		empty_ratings_manager = lib.Hydrus.Ratings.RatingsManager( {} )

		file_viewing_stats_manager = lib.Hydrus.Media.FileViewingStatsManager.STATICGenerateEmptyManager()

		#

		local_hash_has_values = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 1, local_hash_has_values, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, substantial_tags_manager, local_locations_manager, substantial_ratings_manager, file_viewing_stats_manager )

		local_media_has_values = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		other_local_hash_has_values = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 2, other_local_hash_has_values, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, substantial_tags_manager, local_locations_manager, substantial_ratings_manager, file_viewing_stats_manager )

		other_local_media_has_values = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		local_hash_empty = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 3, local_hash_empty, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, empty_tags_manager, local_locations_manager, empty_ratings_manager, file_viewing_stats_manager )

		local_media_empty = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		trashed_hash_empty = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 4, trashed_hash_empty, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, empty_tags_manager, trash_locations_manager, empty_ratings_manager, file_viewing_stats_manager )

		trashed_media_empty = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		deleted_hash_empty = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 5, deleted_hash_empty, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, empty_tags_manager, deleted_locations_manager, empty_ratings_manager, file_viewing_stats_manager )

		deleted_media_empty = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		one_hash = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 6, one_hash, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, one_tags_manager, local_locations_manager, one_ratings_manager, file_viewing_stats_manager )

		one_media = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		two_hash = lib.Hydrus.Data.GenerateKey()

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 7, two_hash, size, mime, width, height, duration, num_frames, num_words )

		media_result = lib.Hydrus.Media.MediaResult( file_info_manager, two_tags_manager, local_locations_manager, two_ratings_manager, file_viewing_stats_manager )

		two_media = lib.Hydrus.Media.MediaSingleton( media_result )

		#

		self._dump_and_load_and_test( duplicate_action_options_delete_and_move, test )
		self._dump_and_load_and_test( duplicate_action_options_copy, test )
		self._dump_and_load_and_test( duplicate_action_options_merge, test )

		#

		def assertSCUEqual( one, two ):
			self.assertEqual( TC.ConvertServiceKeysToContentUpdatesToComparable( one ), TC.ConvertServiceKeysToContentUpdatesToComparable( two ) )

		file_deletion_reason = 'test delete'

		#

		result = duplicate_action_options_delete_and_move.ProcessPairIntoContentUpdates( local_media_has_values, local_media_empty, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_FILE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_DELETE, { local_hash_empty }, reason = file_deletion_reason ) ]

		assertSCUEqual( result, scu )

		#

		result = duplicate_action_options_delete_and_move.ProcessPairIntoContentUpdates( local_media_has_values, trashed_media_empty, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.TRASH_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_DELETE, { trashed_hash_empty }, reason = file_deletion_reason ) ]

		assertSCUEqual( result, scu )

		#

		result = duplicate_action_options_delete_and_move.ProcessPairIntoContentUpdates( local_media_has_values, deleted_media_empty, file_deletion_reason = file_deletion_reason )

		self.assertEqual( result, {} )

		#

		result = duplicate_action_options_delete_and_move.ProcessPairIntoContentUpdates( local_media_has_values, other_local_media_has_values, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_TAG_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_DELETE, ( 'test tag', { other_local_hash_has_values } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_DELETE, ( 'series:namespaced test tag', { other_local_hash_has_values } ) ) ]
		scu[ TC.LOCAL_RATING_LIKE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( None, { other_local_hash_has_values } ) ) ]
		scu[ TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( None, { other_local_hash_has_values } ) ) ]
		scu[ CC.LOCAL_FILE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_DELETE, { other_local_hash_has_values }, reason = file_deletion_reason ) ]

		assertSCUEqual( result, scu )

		#

		result = duplicate_action_options_delete_and_move.ProcessPairIntoContentUpdates( local_media_empty, other_local_media_has_values, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_TAG_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'test tag', { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'series:namespaced test tag', { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_DELETE, ( 'test tag', { other_local_hash_has_values } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_DELETE, ( 'series:namespaced test tag', { other_local_hash_has_values } ) ) ]
		scu[ TC.LOCAL_RATING_LIKE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 1.0, { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( None, { other_local_hash_has_values } ) ) ]
		scu[ TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 0.8, { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( None, { other_local_hash_has_values } ) ) ]
		scu[ CC.LOCAL_FILE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_DELETE, { other_local_hash_has_values }, reason = file_deletion_reason ) ]

		assertSCUEqual( result, scu )

		#
		#

		result = duplicate_action_options_copy.ProcessPairIntoContentUpdates( local_media_has_values, local_media_empty, file_deletion_reason = file_deletion_reason )

		self.assertEqual( result, {} )

		#

		result = duplicate_action_options_copy.ProcessPairIntoContentUpdates( local_media_empty, other_local_media_has_values, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_TAG_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'test tag', { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'series:namespaced test tag', { local_hash_empty } ) ) ]
		scu[ TC.LOCAL_RATING_LIKE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 1.0, { local_hash_empty } ) ) ]
		scu[ TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 0.8, { local_hash_empty } ) ) ]

		assertSCUEqual( result, scu )

		#
		#

		result = duplicate_action_options_merge.ProcessPairIntoContentUpdates( local_media_has_values, local_media_empty, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_TAG_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'test tag', { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'series:namespaced test tag', { local_hash_empty } ) ) ]
		scu[ TC.LOCAL_RATING_LIKE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 1.0, { local_hash_empty } ) ) ]
		scu[ TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 0.8, { local_hash_empty } ) ) ]

		assertSCUEqual( result, scu )

		#

		result = duplicate_action_options_merge.ProcessPairIntoContentUpdates( local_media_empty, other_local_media_has_values, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_TAG_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'test tag', { local_hash_empty } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'series:namespaced test tag', { local_hash_empty } ) ) ]
		scu[ TC.LOCAL_RATING_LIKE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 1.0, { local_hash_empty } ) ) ]
		scu[ TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 0.8, { local_hash_empty } ) ) ]

		assertSCUEqual( result, scu )

		#

		result = duplicate_action_options_merge.ProcessPairIntoContentUpdates( one_media, two_media, file_deletion_reason = file_deletion_reason )

		scu = {}

		scu[ CC.LOCAL_TAG_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'one', { two_hash } ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'two', { one_hash } ) ) ]
		scu[ TC.LOCAL_RATING_LIKE_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 1.0, { two_hash } ) ) ]
		scu[ TC.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 0.8, { two_hash } ) ) ]

		assertSCUEqual( result, scu )

	def test_HG.SerialisableType["SHORTCUT("] self ):
		def test( obj, dupe_obj ):
			self.assertEqual( dupe_obj.__hash__(), ( dupe_obj._shortcut_type, dupe_obj._shortcut_key, tuple( dupe_obj._modifiers ) ).__hash__() )

			self.assertEqual( obj, dupe_obj )

		shortcuts = []

		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut(), 'f7' ) )

		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, wx.WXK_SPACE, [] ), 'space' ) )
		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, ord( 'a' ), [ CC.SHORTCUT_MODIFIER_CTRL ] ), 'ctrl+a' ) )
		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, ord( 'A' ), [ CC.SHORTCUT_MODIFIER_CTRL ] ), 'ctrl+a' ) )
		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, wx.WXK_HOME, [ CC.SHORTCUT_MODIFIER_ALT, CC.SHORTCUT_MODIFIER_CTRL ] ), 'ctrl+alt+home' ) )

		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_MOUSE, CC.SHORTCUT_MOUSE_LEFT, [] ), 'left-click' ) )
		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_MOUSE, CC.SHORTCUT_MOUSE_MIDDLE, [ CC.SHORTCUT_MODIFIER_CTRL ] ), 'ctrl+middle-click' ) )
		shortcuts.append( ( lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_MOUSE, CC.SHORTCUT_MOUSE_SCROLL_DOWN, [ CC.SHORTCUT_MODIFIER_ALT, CC.SHORTCUT_MODIFIER_SHIFT ] ), 'alt+shift+scroll down' ) )

		for ( shortcut, s ) in shortcuts:
			self._dump_and_load_and_test( shortcut, test )

			self.assertEqual( shortcut.ToString(), s )

	def test_HG.SerialisableType["SHORTCUTS("] self ):
		def test( obj, dupe_obj ):
			for ( shortcut, command ) in obj:
				self.assertEqual( dupe_obj.GetCommand( shortcut ).GetData(), command.GetData() )

		default_shortcuts = lib.Hydrus.Defaults.GetDefaultShortcuts()

		for shortcuts in default_shortcuts:
			self._dump_and_load_and_test( shortcuts, test )

		command_1 = lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_SIMPLE, 'archive_file' )
		command_2 = lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_CONTENT, ( lib.Hydrus.Data.GenerateKey(), HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_FLIP, 'test' ) )
		command_3 = lib.Hydrus.Data.Local.ApplicationCommand( CC.APPLICATION_COMMAND_TYPE_CONTENT, ( CC.LOCAL_TAG_SERVICE_KEY, HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_FLIP, 'test' ) )

		k_shortcut_1 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, wx.WXK_SPACE, [] )
		k_shortcut_2 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, ord( 'a' ), [ CC.SHORTCUT_MODIFIER_CTRL ] )
		k_shortcut_3 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, ord( 'A' ), [ CC.SHORTCUT_MODIFIER_CTRL ] )
		k_shortcut_4 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_KEYBOARD, wx.WXK_HOME, [ CC.SHORTCUT_MODIFIER_ALT, CC.SHORTCUT_MODIFIER_CTRL ] )

		m_shortcut_1 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_MOUSE, CC.SHORTCUT_MOUSE_LEFT, [] )
		m_shortcut_2 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_MOUSE, CC.SHORTCUT_MOUSE_MIDDLE, [ CC.SHORTCUT_MODIFIER_CTRL ] )
		m_shortcut_3 = lib.GUI.Shortcuts.Shortcut( CC.SHORTCUT_TYPE_MOUSE, CC.SHORTCUT_MOUSE_SCROLL_DOWN, [ CC.SHORTCUT_MODIFIER_ALT, CC.SHORTCUT_MODIFIER_SHIFT ] )

		shortcuts = lib.GUI.Shortcuts.Shortcuts( 'test' )

		shortcuts.SetCommand( k_shortcut_1, command_1 )
		shortcuts.SetCommand( k_shortcut_2, command_2 )
		shortcuts.SetCommand( k_shortcut_3, command_2 )
		shortcuts.SetCommand( k_shortcut_4, command_3 )

		shortcuts.SetCommand( m_shortcut_1, command_1 )
		shortcuts.SetCommand( m_shortcut_2, command_2 )
		shortcuts.SetCommand( m_shortcut_3, command_3 )

		self._dump_and_load_and_test( shortcuts, test )

		self.assertEqual( shortcuts.GetCommand( k_shortcut_1 ).GetData(), command_1.GetData() )

		shortcuts.SetCommand( k_shortcut_1, command_3 )

		self.assertEqual( shortcuts.GetCommand( k_shortcut_1 ).GetData(), command_3.GetData() )

	def test_HG.SerialisableType["SUBSCRIPTION("] self ):
		def test( obj, dupe_obj ):
			self.assertEqual( obj.GetName(), dupe_obj.GetName() )

			self.assertEqual( obj._gug_key_and_name, dupe_obj._gug_key_and_name )
			self.assertEqual( len( obj._queries ), len( dupe_obj._queries ) )
			self.assertEqual( obj._initial_file_limit, dupe_obj._initial_file_limit )
			self.assertEqual( obj._periodic_file_limit, dupe_obj._periodic_file_limit )
			self.assertEqual( obj._paused, dupe_obj._paused )

			self.assertEqual( obj._file_import_options.GetSerialisableTuple(), dupe_obj._file_import_options.GetSerialisableTuple() )
			self.assertEqual( obj._tag_import_options.GetSerialisableTuple(), dupe_obj._tag_import_options.GetSerialisableTuple() )

			self.assertEqual( obj._no_work_until, dupe_obj._no_work_until )

		sub = lib.Actions.Import.Subscriptions.Subscription( 'test sub' )

		self._dump_and_load_and_test( sub, test )

		gug_key_and_name = ( lib.Hydrus.Data.GenerateKey(), 'muh test gug' )
		queries = [ lib.Actions.Import.Subscriptions.SubscriptionQuery( 'test query' ), lib.Actions.Import.Subscriptions.SubscriptionQuery( 'test query 2' ) ]
		checker_options = lib.Actions.Import.Options.CheckerOptions()
		initial_file_limit = 100
		periodic_file_limit = 50
		paused = False

		file_import_options = lib.Actions.Import.Options.FileImportOptions()

		service_tag_import_options = lib.Actions.Import.Options.ServiceTagImportOptions( get_tags = False, additional_tags = { 'test additional tag', 'and another' } )

		tag_import_options = lib.Actions.Import.Options.TagImportOptions( service_keys_to_service_tag_import_options = { lib.Hydrus.Data.GenerateKey() : service_tag_import_options } )

		no_work_until = lib.Hydrus.Data.GetNow() - 86400 * 20

		sub.SetTuple( gug_key_and_name, queries, checker_options, initial_file_limit, periodic_file_limit, paused, file_import_options, tag_import_options, no_work_until )

		self.assertEqual( sub.GetGUGKeyAndName(), gug_key_and_name )
		self.assertEqual( sub.GetTagImportOptions(), tag_import_options )
		self.assertEqual( sub.GetQueries(), queries )

		self.assertEqual( sub._paused, False )
		sub.PauseResume()
		self.assertEqual( sub._paused, True )
		sub.PauseResume()
		self.assertEqual( sub._paused, False )

		self._dump_and_load_and_test( sub, test )

	def test_HG.SerialisableType["TAG_FILTER("] self ):
		def test( obj, dupe_obj ):
			self.assertEqual( obj._tag_slices_to_rules, dupe_obj._tag_slices_to_rules )

		tags = set()

		tags.add( 'title:test title' )
		tags.add( 'series:neon genesis evangelion' )
		tags.add( 'series:kill la kill' )
		tags.add( 'smile' )
		tags.add( 'blue eyes' )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile', 'blue eyes', 'title:test title', 'series:neon genesis evangelion', 'series:kill la kill' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( '', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), set() )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( '', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'series:', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'series:neon genesis evangelion', 'series:kill la kill' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( '', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'series:kill la kill', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'series:kill la kill' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( '', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'smile', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile', 'blue eyes' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'series:', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile', 'blue eyes', 'series:neon genesis evangelion', 'series:kill la kill' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( ':', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'series:kill la kill', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile', 'blue eyes', 'series:kill la kill' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( 'series:', CC.FILTER_BLACKLIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile', 'blue eyes', 'title:test title' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( 'series:', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'series:neon genesis evangelion', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'smile', 'blue eyes', 'title:test title', 'series:neon genesis evangelion' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( '', CC.FILTER_BLACKLIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'title:test title', 'series:neon genesis evangelion', 'series:kill la kill' } )

		#

		tag_filter = lib.Hydrus.Tags.Local.TagFilter()

		tag_filter.SetRule( '', CC.FILTER_BLACKLIST )
		tag_filter.SetRule( 'blue eyes', CC.FILTER_WHITELIST )

		self._dump_and_load_and_test( tag_filter, test )

		self.assertEqual( tag_filter.Filter( tags ), { 'title:test title', 'series:neon genesis evangelion', 'series:kill la kill', 'blue eyes' } )

license: wtfpl


code:
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.API.Local
import lib.Hydrus.Server.Local
import lib.Hydrus.Media
import lib.Hydrus.Ratings
import lib.Hydrus.Services.Local
import lib.Hydrus.Tags.Local
import hashlib
import http.client
import lib.Hydrus.Constants as HC
import lib.Hydrus.Encryption
import lib.Networking
import lib.Hydrus.Paths
import lib.Hydrus.Server
import lib.Hydrus.Server.Resources
import lib.Hydrus.Text
import json
import os
import random
import lib.Hydrus.Files.Remote
import lib.Hydrus.Server
import ssl
from . import TestController
import time
import unittest
import urllib
from twisted.internet import reactor
#from twisted.internet.endpoints import TCP4ClientEndpoint, connectProtocol
from twisted.internet.defer import deferredGenerator, waitForDeferred
import twisted.internet.ssl
import lib.Hydrus.Data
import lib.Hydrus.Globals as HG

with open( os.path.join( HC.STATIC_DIR, 'hydrus.png' ), 'rb' ) as f:
	EXAMPLE_FILE = f.read()

with open( os.path.join( HC.STATIC_DIR, 'hydrus_small.png' ), 'rb' ) as f:
	EXAMPLE_THUMBNAIL = f.read()

class TestServer( unittest.TestCase ):
	@classmethod

	def setUpClass( cls ):
		cls._access_key = lib.Hydrus.Data.GenerateKey()

		cls._serverside_file_service = lib.Networking.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.FILE_REPOSITORY, 'file repo', HC.DEFAULT_SERVICE_PORT + 1 )
		cls._serverside_tag_service = lib.Networking.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.TAG_REPOSITORY, 'tag repo', HC.DEFAULT_SERVICE_PORT )
		cls._serverside_admin_service = lib.Networking.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.SERVER_ADMIN, 'server admin', HC.DEFAULT_SERVER_ADMIN_PORT )

		cls._clientside_file_service = lib.Hydrus.Services.Local.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.FILE_REPOSITORY, 'file repo' )
		cls._clientside_tag_service = lib.Hydrus.Services.Local.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.TAG_REPOSITORY, 'tag repo' )
		cls._clientside_admin_service = lib.Hydrus.Services.Local.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.SERVER_ADMIN, 'server admin' )

		cls._clientside_file_service.SetCredentials( lib.Networking.Credentials( '127.0.0.1', HC.DEFAULT_SERVICE_PORT + 1, cls._access_key ) )
		cls._clientside_tag_service.SetCredentials( lib.Networking.Credentials( '127.0.0.1', HC.DEFAULT_SERVICE_PORT, cls._access_key ) )
		cls._clientside_admin_service.SetCredentials( lib.Networking.Credentials( '127.0.0.1', HC.DEFAULT_SERVER_ADMIN_PORT, cls._access_key ) )

		cls._local_booru = lib.Hydrus.Services.Local.GenerateService( CC.LOCAL_BOORU_SERVICE_KEY, HC.LOCAL_BOORU, 'local booru' )

		services_manager = HG.test_controller.services_manager

		services_manager._keys_to_services[ cls._clientside_file_service.GetServiceKey() ] = cls._clientside_file_service
		services_manager._keys_to_services[ cls._clientside_tag_service.GetServiceKey() ] = cls._clientside_tag_service
		services_manager._keys_to_services[ cls._clientside_admin_service.GetServiceKey() ] = cls._clientside_admin_service

		permissions = [ HC.GET_DATA, HC.POST_DATA, HC.POST_PETITIONS, HC.RESOLVE_PETITIONS, HC.MANAGE_USERS, HC.GENERAL_ADMIN, HC.EDIT_SERVICES ]

		account_key = lib.Hydrus.Data.GenerateKey()
		account_type = lib.Networking.AccountType.GenerateAdminAccountType( HC.SERVER_ADMIN )
		created = lib.Hydrus.Data.GetNow() - 100000
		expires = None

		cls._account = lib.Networking.Account( account_key, account_type, created, expires )

		cls._file_hash = lib.Hydrus.Data.GenerateKey()

		def TWISTEDSetup():
			cls._ssl_cert_path = os.path.join( TestController.DB_DIR, 'server.crt' )
			cls._ssl_key_path = os.path.join( TestController.DB_DIR, 'server.key' )

			# if db test ran, this is still hanging around and read-only, so don't bother to fail overwriting
			if not os.path.exists( cls._ssl_cert_path ):
				lib.Hydrus.Encryption.GenerateOpenSSLCertAndKeyFile( cls._ssl_cert_path, cls._ssl_key_path )

			context_factory = twisted.internet.ssl.DefaultOpenSSLContextFactory( cls._ssl_key_path, cls._ssl_cert_path )

			reactor.listenSSL( HC.DEFAULT_SERVER_ADMIN_PORT, lib.Hydrus.Server.HydrusServiceAdmin( cls._serverside_admin_service ), context_factory )
			reactor.listenSSL( HC.DEFAULT_SERVICE_PORT + 1, lib.Hydrus.Server.HydrusServiceRepositoryFile( cls._serverside_file_service ), context_factory )
			reactor.listenSSL( HC.DEFAULT_SERVICE_PORT, lib.Hydrus.Server.HydrusServiceRepositoryTag( cls._serverside_tag_service ), context_factory )

			reactor.listenTCP( 45866, lib.Hydrus.Server.Local.HydrusServiceBooru( cls._local_booru, allow_non_local_connections = False ) )

		reactor.callFromThread( TWISTEDSetup )

		time.sleep( 1 )

	def _test_basics( self, host, port, https = True ):
		if https:
			context = ssl.SSLContext( ssl.PROTOCOL_SSLv23 )
			context.options |= ssl.OP_NO_SSLv2
			context.options |= ssl.OP_NO_SSLv3

			connection = http.client.HTTPSConnection( host, port, timeout = 10, context = context )

		else:
			connection = http.client.HTTPConnection( host, port, timeout = 10 )

		#

		connection.request( 'GET', '/' )

		response = connection.getresponse()

		data = response.read()

		self.assertEqual( response.status, 200 )

		#

		with open( os.path.join( HC.STATIC_DIR, 'hydrus.ico' ), 'rb' ) as f:
			favicon = f.read()

		connection.request( 'GET', '/favicon.ico' )

		response = connection.getresponse()

		data = response.read()

		self.assertEqual( data, favicon )

	def _test_file_repo( self, service ):
		# file

		path = lib.Hydrus.Files.Remote.GetExpectedFilePath( self._file_hash )

		lib.Hydrus.Paths.MakeSureDirectoryExists( os.path.dirname( path ) )

		with open( path, 'wb' ) as f:
			f.write( EXAMPLE_FILE )

		response = service.Request( HC.GET, 'file', { 'hash' : self._file_hash } )

		self.assertEqual( response, EXAMPLE_FILE )

		try:
			os.remove( path )
		except Exception:
			pass

		path = os.path.join( HC.STATIC_DIR, 'hydrus.png' )

		with open( path, 'rb' ) as f:
			file_bytes = f.read()

		HG.test_controller.ClearWrites( 'file' )

		service.Request( HC.POST, 'file', { 'file' : file_bytes } )

		written = HG.test_controller.GetWrite( 'file' )

		[ ( args, kwargs ) ] = written

		( written_service_key, written_account, written_file_dict ) = args

		self.assertEqual( written_file_dict[ 'hash' ], b'\xadm5\x99\xa6\xc4\x89\xa5u\xeb\x19\xc0&\xfa\xce\x97\xa9\xcdey\xe7G(\xb0\xce\x94\xa6\x01\xd22\xf3\xc3' )
		self.assertEqual( written_file_dict[ 'ip' ], '127.0.0.1' )
		self.assertEqual( written_file_dict[ 'height' ], 200 )
		self.assertEqual( written_file_dict[ 'width' ], 200 )
		self.assertEqual( written_file_dict[ 'mime' ], 2 )
		self.assertEqual( written_file_dict[ 'size' ], 5270 )

		# ip

		( ip, timestamp ) = ( '94.45.87.123', lib.Hydrus.Data.GetNow() - 100000 )

		HG.test_controller.SetRead( 'ip', ( ip, timestamp ) )

		response = service.Request( HC.GET, 'ip', { 'hash' : self._file_hash } )

		self.assertEqual( response[ 'ip' ], ip )
		self.assertEqual( response[ 'timestamp' ], timestamp )

		# thumbnail

		path = lib.Hydrus.Files.Remote.GetExpectedThumbnailPath( self._file_hash )

		lib.Hydrus.Paths.MakeSureDirectoryExists( os.path.dirname( path ) )

		with open( path, 'wb' ) as f:
			f.write( EXAMPLE_THUMBNAIL )

		response = service.Request( HC.GET, 'thumbnail', { 'hash' : self._file_hash } )

		self.assertEqual( response, EXAMPLE_THUMBNAIL )

		try:
			os.remove( path )
		except Exception:
			pass

	def _test_local_booru( self, host, port ):
		#

		connection = http.client.HTTPConnection( host, port, timeout = 10 )

		#

		with open( os.path.join( HC.STATIC_DIR, 'local_booru_style.css' ), 'rb' ) as f:
			css = f.read()

		connection.request( 'GET', '/style.css' )

		response = connection.getresponse()

		data = response.read()

		self.assertEqual( data, css )

		#

		share_key = lib.Hydrus.Data.GenerateKey()
		hashes = [ lib.Hydrus.Data.GenerateKey() for i in range( 5 ) ]

		client_files_default = os.path.join( TestController.DB_DIR, 'client_files' )

		hash_encoded = hashes[0].hex()

		prefix = hash_encoded[:2]

		file_path =  os.path.join( client_files_default, 'f' + prefix, hash_encoded + '.jpg' )
		thumbnail_path = os.path.join( client_files_default, 't' + prefix, hash_encoded + '.thumbnail' )

		with open( file_path, 'wb' ) as f:
			f.write( EXAMPLE_FILE )

		with open( thumbnail_path, 'wb' ) as f:
			f.write( EXAMPLE_THUMBNAIL )

		local_booru_manager = HG.client_controller.local_booru_manager

		#

		self._test_local_booru_requests( connection, share_key, hashes[0], 404 )

		#

		info = {}
		info[ 'name' ] = 'name'
		info[ 'text' ] = 'text'
		info[ 'timeout' ] = 0
		info[ 'hashes' ] = hashes

		file_info_manager = lib.Hydrus.Media.FileInfoManager( 1, hashes[0], 500, HC.IMAGE_JPEG, 640, 480 )

		file_viewing_stats_manager = lib.Hydrus.Media.FileViewingStatsManager.STATICGenerateEmptyManager()

		media_results = [ lib.Hydrus.Media.MediaResult( file_info_manager, lib.Hydrus.Media.TagsManager( {} ), lib.Hydrus.Media.LocationsManager( set(), set(), set(), set() ), lib.Hydrus.Ratings.RatingsManager( {} ), file_viewing_stats_manager ) for hash in hashes ]

		HG.test_controller.SetRead( 'local_booru_share_keys', [ share_key ] )
		HG.test_controller.SetRead( 'local_booru_share', info )
		HG.test_controller.SetRead( 'media_results', media_results )

		local_booru_manager.RefreshShares()

		#

		self._test_local_booru_requests( connection, share_key, hashes[0], 403 )

		#

		info[ 'timeout' ] = None
		HG.test_controller.SetRead( 'local_booru_share', info )

		local_booru_manager.RefreshShares()

		#

		self._test_local_booru_requests( connection, share_key, hashes[0], 200 )

		#

		HG.test_controller.SetRead( 'local_booru_share_keys', [] )

		local_booru_manager.RefreshShares()

		#

		self._test_local_booru_requests( connection, share_key, hashes[0], 404 )

	def _test_local_booru_requests( self, connection, share_key, hash, expected_result ):
		requests = []

		requests.append( '/gallery?share_key=' + share_key.hex() )
		requests.append( '/page?share_key=' + share_key.hex() + '&hash=' + hash.hex() )
		requests.append( '/file?share_key=' + share_key.hex() + '&hash=' + hash.hex() )
		requests.append( '/thumbnail?share_key=' + share_key.hex() + '&hash=' + hash.hex() )

		for request in requests:
			connection.request( 'GET', request )

			response = connection.getresponse()

			data = response.read()

			self.assertEqual( response.status, expected_result )

	def _test_repo( self, service ):
		service_key = service.GetServiceKey()

		# num_petitions

		num_petitions = [ [ HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_STATUS_PETITIONED, 23 ], [ HC.CONTENT_TYPE_TAG_PARENTS, HC.CONTENT_STATUS_PENDING, 0 ] ]

		HG.test_controller.SetRead( 'num_petitions', num_petitions )

		response = service.Request( HC.GET, 'num_petitions' )

		self.assertEqual( response[ 'num_petitions' ], num_petitions )

		# petition

		action = HC.CONTENT_UPDATE_PETITION
		petitioner_account = lib.Networking.Account.GenerateUnknownAccount()
		reason = 'it sucks'
		contents = [ lib.Networking.Content( HC.CONTENT_TYPE_FILES, [ lib.Hydrus.Data.GenerateKey() for i in range( 10 ) ] ) ]

		petition = lib.Networking.Petition( action, petitioner_account, reason, contents )

		HG.test_controller.SetRead( 'petition', petition )

		response = service.Request( HC.GET, 'petition', { 'content_type' : HC.CONTENT_TYPE_FILES, 'status' : HC.CONTENT_UPDATE_PETITION } )

		self.assertEqual( response[ 'petition' ].GetSerialisableTuple(), petition.GetSerialisableTuple() )

		# definitions

		definitions_update = lib.Networking.DefinitionsUpdate()

		for i in range( 100, 200 ):
			definitions_update.AddRow( ( HC.DEFINITIONS_TYPE_TAGS, i, 'series:test ' + str( i ) ) )
			definitions_update.AddRow( ( HC.DEFINITIONS_TYPE_HASHES, i + 500, lib.Hydrus.Data.GenerateKey() ) )

		definitions_update_network_bytes = definitions_update.DumpToNetworkBytes()

		definitions_update_hash = hashlib.sha256( definitions_update_network_bytes ).digest()

		path = lib.Hydrus.Files.Remote.GetExpectedFilePath( definitions_update_hash )

		lib.Hydrus.Paths.MakeSureDirectoryExists( path )

		with open( path, 'wb' ) as f:
			f.write( definitions_update_network_bytes )

		response = service.Request( HC.GET, 'update', { 'update_hash' : definitions_update_hash } )

		try:
			os.remove( path )
		except Exception:
			pass

		self.assertEqual( response, definitions_update_network_bytes )

		# content

		rows = [ ( random.randint( 100, 1000 ), [ random.randint( 100, 1000 ) for i in range( 50 ) ] ) for j in range( 20 ) ]

		content_update = lib.Networking.ContentUpdate()

		for row in rows:
			content_update.AddRow( ( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, row ) )

		content_update_network_bytes = content_update.DumpToNetworkBytes()

		content_update_hash = hashlib.sha256( content_update_network_bytes ).digest()

		path = lib.Hydrus.Files.Remote.GetExpectedFilePath( content_update_hash )

		with open( path, 'wb' ) as f:
			f.write( content_update_network_bytes )

		response = service.Request( HC.GET, 'update', { 'update_hash' : content_update_hash } )

		try:
			os.remove( path )
		except Exception:
			pass

		self.assertEqual( response, content_update_network_bytes )

		# metadata

		metadata = lib.Networking.Metadata()

		metadata.AppendUpdate( [ definitions_update_hash, content_update_hash ], lib.Hydrus.Data.GetNow() - 101000, lib.Hydrus.Data.GetNow() - 1000, lib.Hydrus.Data.GetNow() + 100000 )

		service._metadata = metadata

		response = service.Request( HC.GET, 'metadata_slice', { 'since' : 0 } )

		self.assertEqual( response[ 'metadata_slice' ].GetSerialisableTuple(), metadata.GetSerialisableTuple() )

		# post content

		raise NotImplementedError()
		'''
		update = lib.Hydrus.Data.ClientToServerContentUpdatePackage( {}, hash_ids_to_hashes )

		HG.test_controller.ClearWrites( 'update' )

		service.Request( HC.POST, 'content_update_package', { 'update' : update } )

		written = HG.test_controller.GetWrite( 'update' )

		[ ( args, kwargs ) ] = written

		( written_service_key, written_account, written_update ) = args

		self.assertEqual( update.GetHashes(), written_update.GetHashes() )
		'''

	def _test_restricted( self, service ):
		# access_key

		registration_key = lib.Hydrus.Data.GenerateKey()

		HG.test_controller.SetRead( 'access_key', self._access_key )

		response = service.Request( HC.GET, 'access_key', { 'registration_key' : registration_key } )

		self.assertEqual( response[ 'access_key' ], self._access_key )

		# set up session

		last_error = 0

		account = self._account

		HG.test_controller.SetRead( 'service', service )

		HG.test_controller.SetRead( 'account_key_from_access_key', lib.Hydrus.Data.GenerateKey() )
		HG.test_controller.SetRead( 'account', self._account )

		# account

		response = service.Request( HC.GET, 'account' )

		self.assertEqual( repr( response[ 'account' ] ), repr( self._account ) )

		# account_info

		account_info = { 'message' : 'hello' }

		HG.test_controller.SetRead( 'account_info', account_info )
		HG.test_controller.SetRead( 'account_key_from_identifier', lib.Hydrus.Data.GenerateKey() )

		response = service.Request( HC.GET, 'account_info', { 'subject_account_key' : lib.Hydrus.Data.GenerateKey() } )

		self.assertEqual( response[ 'account_info' ], account_info )

		# this not working for now, screw it m8

		#response = service.Request( HC.GET, 'account_info', { 'subject_hash' : lib.Hydrus.Data.GenerateKey() } )

		#self.assertEqual( response[ 'account_info' ], account_info )

		#response = service.Request( HC.GET, 'account_info', { 'subject_hash' : lib.Hydrus.Data.GenerateKey(), 'subject_tag' : 'hello' } )

		#self.assertEqual( response[ 'account_info' ], account_info )

		# account_types

		account_types = [ lib.Networking.AccountType.GenerateAdminAccountType( service.GetServiceType() ) ]

		HG.test_controller.SetRead( 'account_types', account_types )

		HG.test_controller.ClearWrites( 'account_types' )

		response = service.Request( HC.GET, 'account_types' )

		self.assertEqual( response[ 'account_types' ][0].GetAccountTypeKey(), account_types[0].GetAccountTypeKey() )

		service.Request( HC.POST, 'account_types', { 'account_types' : account_types, 'deletee_account_type_keys_to_new_account_type_keys' : {} } )

		written = HG.test_controller.GetWrite( 'account_types' )

		[ ( args, kwargs ) ] = written

		( written_service_key, written_account, written_account_types, written_deletee_account_type_keys_to_new_account_type_keys ) = args

		self.assertEqual( written_account_types[0].GetAccountTypeKey(), account_types[0].GetAccountTypeKey() )
		self.assertEqual( written_deletee_account_type_keys_to_new_account_type_keys, {} )

		# registration_keys

		registration_key = lib.Hydrus.Data.GenerateKey()

		HG.test_controller.SetRead( 'registration_keys', [ registration_key ] )

		response = service.Request( HC.GET, 'registration_keys', { 'num' : 1, 'account_type_key' : os.urandom( 32 ), 'expires' : lib.Hydrus.Data.GetNow() + 1200 } )

		self.assertEqual( response[ 'registration_keys' ], [ registration_key ] )

	def _test_server_admin( self, service ):
		# init

		access_key = lib.Hydrus.Data.GenerateKey()

		HG.test_controller.SetRead( 'access_key', access_key )

		response = service.Request( HC.GET, 'access_key', { 'registration_key' : b'init' } )

		self.assertEqual( response[ 'access_key' ], access_key )

		#

		## backup

		response = service.Request( HC.POST, 'backup' )

		#

		# add some new services info

	def _test_tag_repo( self, service ):
		pass

	def test_repository_file( self ):
		host = '127.0.0.1'
		port = HC.DEFAULT_SERVICE_PORT

		self._test_basics( host, port )
		self._test_restricted( self._clientside_file_service )
		# broke since service rewrite
		#self._test_repo( self._clientside_file_service )
		#self._test_file_repo( self._clientside_file_service )

	def test_repository_tag( self ):
		host = '127.0.0.1'
		port = HC.DEFAULT_SERVICE_PORT + 1

		self._test_basics( host, port )
		self._test_restricted( self._clientside_tag_service )
		# broke since service rewrite
		#self._test_repo( self._clientside_tag_service )
		#self._test_tag_repo( self._clientside_tag_service )

	def test_server_admin( self ):
		host = '127.0.0.1'
		port = HC.DEFAULT_SERVER_ADMIN_PORT

		self._test_basics( host, port )
		self._test_restricted( self._clientside_admin_service )
		self._test_server_admin( self._clientside_admin_service )

	def test_local_booru( self ):
		host = '127.0.0.1'
		port = 45866

		self._test_basics( host, port, https = False )
		self._test_local_booru( host, port )

	'''

class TestAMP( unittest.TestCase ):
	@classmethod

	def setUpClass( cls ):
		cls._alice = lib.Hydrus.Data.GenerateKey()
		cls._bob = lib.Hydrus.Data.GenerateKey()

		cls._server_port = HC.DEFAULT_SERVICE_PORT + 10

		cls._service_key = lib.Hydrus.Data.GenerateKey()

		def TWISTEDSetup():
			cls._factory = lib.Hydrus.Server.MessagingServiceFactory( cls._service_key )

			reactor.listenTCP( cls._server_port, cls._factory )

		reactor.callFromThread( TWISTEDSetup )

		time.sleep( 1 )

	def _get_deferred_result( self, deferred ):
		def err( failure ):
			failure.trap( Exception )

			return failure.type( failure.value )

		deferred.addErrback( err )

		before = time.time()

		while not deferred.called:
			time.sleep( 0.1 )

			if time.time() - before > 10:
				raise Exception( 'Trying to get deferred timed out!' )

		result = deferred.result

		if issubclass( type( result ), Exception ):
			raise result

		return result

	def _get_client_protocol( self ):
		point = TCP4ClientEndpoint( reactor, '127.0.0.1', self._server_port )

		deferred = connectProtocol( point, lib.Hydrus.Server.AMP.MessagingClientProtocol() )

		protocol = self._get_deferred_result( deferred )

		return protocol

	def _make_persistent_connection( self, protocol, access_key, name ):
		identifier = hashlib.sha256( access_key ).digest()

		HC.app.SetRead( 'im_identifier', identifier )

		permissions = [ HC.GET_DATA, HC.POST_DATA, HC.POST_PETITIONS, HC.RESOLVE_PETITIONS, HC.MANAGE_USERS, HC.GENERAL_ADMIN, HC.EDIT_SERVICES ]

		account_key = lib.Hydrus.Data.GenerateKey()
		account_type = HC.AccountType( 'account', permissions, ( None, None ) )
		created = HC.GetNow() - 100000
		expires = None
		used_bytes = 0
		used_requests = 0

		account = HC.Account( account_key, account_type, created, expires, used_bytes, used_requests )

		HC.app.SetRead( 'account_key_from_access_key', lib.Hydrus.Data.GenerateKey() )
		HC.app.SetRead( 'account', account )

		deferred = protocol.callRemote( lib.Hydrus.Server.AMP.IMSessionKey, access_key = access_key, name = name )

		result = self._get_deferred_result( deferred )

		session_key = result[ 'session_key' ]

		deferred = protocol.callRemote( lib.Hydrus.Server.AMP.IMLoginPersistent, network_version = HC.NETWORK_VERSION, session_key = session_key )

		result = self._get_deferred_result( deferred )

		self.assertEqual( result, {} )

	def _make_temporary_connection( self, protocol, identifier, name ):
		deferred = protocol.callRemote( lib.Hydrus.Server.AMP.IMLoginTemporary, network_version = HC.NETWORK_VERSION, identifier = identifier, name = name )

		result = self._get_deferred_result( deferred )

		self.assertEqual( result, {} )

	def test_connections( self ):
		persistent_protocol = self._get_client_protocol()
		persistent_access_key = lib.Hydrus.Data.GenerateKey()
		persistent_identifier = hashlib.sha256( persistent_access_key ).digest()
		persistent_name = 'persistent'

		self._make_persistent_connection( persistent_protocol, persistent_access_key, persistent_name )

		self.assertIn( persistent_identifier, self._factory._persistent_connections )
		self.assertIn( persistent_name, self._factory._persistent_connections[ persistent_identifier ] )

		temp_protocol_1 = self._get_client_protocol()
		temp_protocol_2 = self._get_client_protocol()
		temp_name_1 = 'temp_1'
		temp_identifier = lib.Hydrus.Data.GenerateKey()
		temp_name_2 = 'temp_2'

		self._make_temporary_connection( temp_protocol_1, temp_identifier, temp_name_1 )
		self._make_temporary_connection( temp_protocol_2, temp_identifier, temp_name_2 )

		self.assertIn( temp_identifier, self._factory._temporary_connections )
		self.assertIn( temp_name_1, self._factory._temporary_connections[ temp_identifier ] )
		self.assertIn( temp_name_2, self._factory._temporary_connections[ temp_identifier ] )

	def test_status( self ):
		# some of this is UDP, so get that working!

		# add two bobs

		# ask for status of the bobs
		# test that we get both, online

		# now disconnect a bob
		# ask for bob status
		# test that we only have one bob

		# now disconnect other bob
		# repeat for nothing

		pass

	def test_message( self ):
		persistent_protocol = self._get_client_protocol()
		persistent_access_key = lib.Hydrus.Data.GenerateKey()
		persistent_identifier = hashlib.sha256( persistent_access_key ).digest()
		persistent_name = 'persistent'

		self._make_persistent_connection( persistent_protocol, persistent_access_key, persistent_name )

		temp_protocol = self._get_client_protocol()
		temp_identifier = lib.Hydrus.Data.GenerateKey()
		temp_name = 'temp'

		self._make_temporary_connection( temp_protocol, temp_identifier, temp_name )

		#

		HC.pubsub.ClearPubSubs()

		message = 'hello temp'

		deferred = persistent_protocol.callRemote( lib.Hydrus.Server.AMP.IMMessageServer, identifier_to = temp_identifier, name_to = temp_name, message = message )

		result = self._get_deferred_result( deferred )

		self.assertEqual( result, {} )

		result = HC.pubsub.GetPubSubs( 'im_message_received' )

		[ ( args, kwargs ) ] = result

		self.assertEqual( args, ( persistent_identifier, persistent_name, temp_identifier, temp_name, message ) )

		#

		HC.pubsub.ClearPubSubs()

		message = 'hello persistent'

		deferred = temp_protocol.callRemote( lib.Hydrus.Server.AMP.IMMessageServer, identifier_to = persistent_identifier, name_to = persistent_name, message = message )

		result = self._get_deferred_result( deferred )

		self.assertEqual( result, {} )

		result = HC.pubsub.GetPubSubs( 'im_message_received' )

		[ ( args, kwargs ) ] = result

		self.assertEqual( args, ( temp_identifier, temp_name, persistent_identifier, persistent_name, message ) )
		'''


code:
import lib.Hydrus.Constants.Local as CC
import collections
import lib.Hydrus.Constants as HC
import lib.Hydrus.Exceptions
import lib.Networking
import lib.Hydrus.Sessions
import os
import unittest
import lib.Hydrus.Data
import lib.Hydrus.Globals as HG

class TestSessions( unittest.TestCase ):
	def test_server( self ):
		discard = HG.test_controller.GetWrite( 'session' ) # just to discard gumph from testserver

		session_key_1 = lib.Hydrus.Data.GenerateKey()
		service_key = lib.Hydrus.Data.GenerateKey()

		permissions = [ HC.GET_DATA, HC.POST_DATA, HC.POST_PETITIONS, HC.RESOLVE_PETITIONS, HC.MANAGE_USERS, HC.GENERAL_ADMIN, HC.EDIT_SERVICES ]

		access_key = lib.Hydrus.Data.GenerateKey()
		account_key = lib.Hydrus.Data.GenerateKey()
		account_type = lib.Networking.AccountType.GenerateAdminAccountType( HC.SERVER_ADMIN )
		created = lib.Hydrus.Data.GetNow() - 100000
		expires = lib.Hydrus.Data.GetNow() + 300

		account = lib.Networking.Account( account_key, account_type, created, expires )

		expires = lib.Hydrus.Data.GetNow() - 10

		HG.test_controller.SetRead( 'sessions', [ ( session_key_1, service_key, account, expires ) ] )

		session_manager = lib.Hydrus.Sessions.HydrusSessionManagerServer()

		with self.assertRaises( lib.Hydrus.Exceptions.SessionException ):
			session_manager.GetAccount( service_key, session_key_1 )

		# test fetching a session already in db, after bootup

		expires = lib.Hydrus.Data.GetNow() + 300

		HG.test_controller.SetRead( 'sessions', [ ( session_key_1, service_key, account, expires ) ] )

		session_manager = lib.Hydrus.Sessions.HydrusSessionManagerServer()

		read_account = session_manager.GetAccount( service_key, session_key_1 )

		self.assertIs( read_account, account )

		# test adding a session

		HG.test_controller.ClearWrites( 'session' )

		expires = lib.Hydrus.Data.GetNow() + 300

		account_key_2 = lib.Hydrus.Data.GenerateKey()

		account_2 = lib.Networking.Account( account_key_2, account_type, created, expires )

		HG.test_controller.SetRead( 'account_key_from_access_key', account_key_2 )
		HG.test_controller.SetRead( 'account', account_2 )

		( session_key_2, expires_2 ) = session_manager.AddSession( service_key, access_key )

		[ ( args, kwargs ) ] = HG.test_controller.GetWrite( 'session' )

		( written_session_key, written_service_key, written_account_key, written_expires ) = args

		self.assertEqual( ( session_key_2, service_key, account_key_2, expires_2 ), ( written_session_key, written_service_key, written_account_key, written_expires ) )

		read_account = session_manager.GetAccount( service_key, session_key_2 )

		self.assertIs( read_account, account_2 )

		# test adding a new session for an account already in the manager

		HG.test_controller.SetRead( 'account_key_from_access_key', account_key )
		HG.test_controller.SetRead( 'account', account )

		( session_key_3, expires_3 ) = session_manager.AddSession( service_key, access_key )

		[ ( args, kwargs ) ] = HG.test_controller.GetWrite( 'session' )

		( written_session_key, written_service_key, written_account_key, written_expires ) = args

		self.assertEqual( ( session_key_3, service_key, account_key, expires_3 ), ( written_session_key, written_service_key, written_account_key, written_expires ) )

		read_account = session_manager.GetAccount( service_key, session_key_3 )

		self.assertIs( read_account, account )

		read_account_original = session_manager.GetAccount( service_key, session_key_1 )

		self.assertIs( read_account, read_account_original )

		# test individual account refresh

		expires = lib.Hydrus.Data.GetNow() + 300

		updated_account = lib.Networking.Account( account_key, account_type, created, expires )

		HG.test_controller.SetRead( 'account', updated_account )

		session_manager.RefreshAccounts( service_key, [ account_key ] )

		read_account = session_manager.GetAccount( service_key, session_key_1 )

		self.assertIs( read_account, updated_account )

		read_account = session_manager.GetAccount( service_key, session_key_3 )

		self.assertIs( read_account, updated_account )

		# test all account refresh

		expires = lib.Hydrus.Data.GetNow() + 300

		updated_account_2 = lib.Networking.Account( account_key, account_type, created, expires )

		HG.test_controller.SetRead( 'sessions', [ ( session_key_1, service_key, updated_account_2, expires ), ( session_key_2, service_key, account_2, expires ), ( session_key_3, service_key, updated_account_2, expires ) ] )

		session_manager.RefreshAllAccounts()

		read_account = session_manager.GetAccount( service_key, session_key_1 )

		self.assertIs( read_account, updated_account_2 )

		read_account = session_manager.GetAccount( service_key, session_key_2 )

		self.assertIs( read_account, account_2 )

		read_account = session_manager.GetAccount( service_key, session_key_3 )

		self.assertIs( read_account, updated_account_2 )