code:
import lib.Hydrus.Constants.Local as CC
import lib.Networking
import lib.Networking.Bandwidth
import lib.Networking.Contexts
import lib.Networking.Domain
import lib.Networking.Jobs
import lib.Networking.Login
import lib.Networking.Sessions
import lib.Hydrus.Services.Local
import collections
import lib.Hydrus.Constants as HC
import lib.Hydrus.Data
import lib.Hydrus.Exceptions
import lib.Networking
import os
from . import TestController
import threading
import time
import unittest
import lib.Hydrus.Globals as HG
from httmock import all_requests, urlmatch, HTTMock, response
from mock import patch

# some gumpf
GOOD_RESPONSE = bytes( range( 256 ) )

# 256KB of gumpf
LONG_GOOD_RESPONSE = GOOD_RESPONSE * 4 * 256

BAD_RESPONSE = b'500, it done broke'

@all_requests

def catch_all( url, request ):
	raise Exception( 'An unexpected request for ' + url + ' came through in testing.' )

MOCK_DOMAIN = 'wew.lad'
MOCK_SUBDOMAIN = 'top.wew.lad'
MOCK_URL = 'https://wew.lad/folder/request&key1=value1&key2=value2'
MOCK_SUBURL = 'https://top.wew.lad/folder2/request&key1=value1&key2=value2'

MOCK_HYDRUS_SERVICE_KEY = lib.Hydrus.Data.GenerateKey()
MOCK_HYDRUS_ADDRESS = '123.45.67.89'
MOCK_HYDRUS_DOMAIN = '123.45.67.89:45871'
MOCK_HYDRUS_URL = 'https://123.45.67.89:45871/muh_hydrus_command'

@urlmatch( netloc = 'wew.lad' )

def catch_wew_error( url, request ):
	return { 'status_code' : 500, 'reason' : 'Internal Server Error', 'content' : BAD_RESPONSE }

@urlmatch( netloc = 'wew.lad' )

def catch_wew_ok( url, request ):
	return GOOD_RESPONSE

@urlmatch( netloc = MOCK_HYDRUS_ADDRESS )

def catch_hydrus_error( url, request ):
	return response( 500, BAD_RESPONSE, { 'Server' : HC.service_string_lookup[ HC.TAG_REPOSITORY ] + '/' + str( HC.NETWORK_VERSION ) }, 'Internal Server Error' )

@urlmatch( netloc = MOCK_HYDRUS_ADDRESS )

def catch_hydrus_ok( url, request ):
	return response( 200, GOOD_RESPONSE, { 'Server' : HC.service_string_lookup[ HC.TAG_REPOSITORY ] + '/' + str( HC.NETWORK_VERSION ) }, 'OK' )

class TestBandwidthManager( unittest.TestCase ):
	def test_can_start( self ):
		EMPTY_RULES = lib.Networking.BandwidthRules()

		PERMISSIVE_DATA_RULES = lib.Networking.BandwidthRules()

		PERMISSIVE_DATA_RULES.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 1048576 )

		PERMISSIVE_REQUEST_RULES = lib.Networking.BandwidthRules()

		PERMISSIVE_REQUEST_RULES.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, None, 10000 )

		RESTRICTIVE_DATA_RULES = lib.Networking.BandwidthRules()

		RESTRICTIVE_DATA_RULES.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 10 )

		RESTRICTIVE_REQUEST_RULES = lib.Networking.BandwidthRules()

		RESTRICTIVE_REQUEST_RULES.AddRule( HC.BANDWIDTH_TYPE_REQUESTS, None, 1 )

		DOMAIN_NETWORK_CONTEXT = lib.Networking.Contexts.NetworkContext( CC.NETWORK_CONTEXT_DOMAIN, MOCK_DOMAIN )
		SUBDOMAIN_NETWORK_CONTEXT = lib.Networking.Contexts.NetworkContext( CC.NETWORK_CONTEXT_DOMAIN, MOCK_SUBDOMAIN )

		GLOBAL_NETWORK_CONTEXTS = [ lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT ]
		DOMAIN_NETWORK_CONTEXTS = [ lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, DOMAIN_NETWORK_CONTEXT ]
		SUBDOMAIN_NETWORK_CONTEXTS = [ lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, DOMAIN_NETWORK_CONTEXT, SUBDOMAIN_NETWORK_CONTEXT ]

		#

		fast_forward = lib.Hydrus.Data.GetNow() + 3600

		with patch.object( HydrusData, 'GetNow', return_value = fast_forward ):
			bm = lib.Networking.Bandwidth.NetworkBandwidthManager()

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			#

			bm.ReportRequestUsed( DOMAIN_NETWORK_CONTEXTS )
			bm.ReportDataUsed( DOMAIN_NETWORK_CONTEXTS, 50 )
			bm.ReportRequestUsed( SUBDOMAIN_NETWORK_CONTEXTS )
			bm.ReportDataUsed( SUBDOMAIN_NETWORK_CONTEXTS, 25 )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			#

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, EMPTY_RULES )
			bm.SetRules( DOMAIN_NETWORK_CONTEXT, EMPTY_RULES )
			bm.SetRules( SUBDOMAIN_NETWORK_CONTEXT, EMPTY_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )
			bm.SetRules( DOMAIN_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )
			bm.SetRules( SUBDOMAIN_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, PERMISSIVE_REQUEST_RULES )
			bm.SetRules( DOMAIN_NETWORK_CONTEXT, PERMISSIVE_REQUEST_RULES )
			bm.SetRules( SUBDOMAIN_NETWORK_CONTEXT, PERMISSIVE_REQUEST_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			#

			bm.SetRules( SUBDOMAIN_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( SUBDOMAIN_NETWORK_CONTEXT, RESTRICTIVE_REQUEST_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( SUBDOMAIN_NETWORK_CONTEXT, PERMISSIVE_REQUEST_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			#

			bm.SetRules( DOMAIN_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( DOMAIN_NETWORK_CONTEXT, RESTRICTIVE_REQUEST_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( DOMAIN_NETWORK_CONTEXT, PERMISSIVE_REQUEST_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			#

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

			self.assertFalse( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, RESTRICTIVE_REQUEST_RULES )

			self.assertFalse( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, PERMISSIVE_REQUEST_RULES )

			self.assertTrue( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertTrue( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

			#

			bm.SetRules( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )
			bm.SetRules( DOMAIN_NETWORK_CONTEXT, RESTRICTIVE_REQUEST_RULES )
			bm.SetRules( DOMAIN_NETWORK_CONTEXT, EMPTY_RULES )

			self.assertFalse( bm.CanStartRequest( GLOBAL_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( DOMAIN_NETWORK_CONTEXTS ) )
			self.assertFalse( bm.CanStartRequest( SUBDOMAIN_NETWORK_CONTEXTS ) )

	def test_can_continue( self ):
		pass

class TestNetworkingEngine( unittest.TestCase ):
	def test_engine_shutdown_app( self ):
		mock_controller = TestController.MockController()
		bandwidth_manager = lib.Networking.Bandwidth.NetworkBandwidthManager()
		session_manager = lib.Networking.Sessions.NetworkSessionManager()
		domain_manager = lib.Networking.Domain.NetworkDomainManager()
		login_manager = lib.Networking.Login.NetworkLoginManager()

		engine = lib.Networking.NetworkEngine( mock_controller, bandwidth_manager, session_manager, domain_manager, login_manager )

		self.assertFalse( engine.IsRunning() )
		self.assertFalse( engine.IsShutdown() )

		mock_controller.CallToThread( engine.MainLoop )

		time.sleep( 0.1 )

		self.assertTrue( engine.IsRunning() )
		self.assertFalse( engine.IsShutdown() )

		mock_controller.model_is_shutdown = True

		engine._new_work_to_do.set()

		time.sleep( 0.1 )

		self.assertFalse( engine.IsRunning() )
		self.assertTrue( engine.IsShutdown() )

	def test_engine_shutdown_manual( self ):
		mock_controller = TestController.MockController()
		bandwidth_manager = lib.Networking.Bandwidth.NetworkBandwidthManager()
		session_manager = lib.Networking.Sessions.NetworkSessionManager()
		domain_manager = lib.Networking.Domain.NetworkDomainManager()
		login_manager = lib.Networking.Login.NetworkLoginManager()

		engine = lib.Networking.NetworkEngine( mock_controller, bandwidth_manager, session_manager, domain_manager, login_manager )

		self.assertFalse( engine.IsRunning() )
		self.assertFalse( engine.IsShutdown() )

		mock_controller.CallToThread( engine.MainLoop )

		time.sleep( 0.1 )

		self.assertTrue( engine.IsRunning() )
		self.assertFalse( engine.IsShutdown() )

		engine.Shutdown()

		time.sleep( 0.1 )

		self.assertFalse( engine.IsRunning() )
		self.assertTrue( engine.IsShutdown() )

	def test_engine_simple_job( self ):
		mock_controller = TestController.MockController()
		bandwidth_manager = lib.Networking.Bandwidth.NetworkBandwidthManager()
		session_manager = lib.Networking.Sessions.NetworkSessionManager()
		domain_manager = lib.Networking.Domain.NetworkDomainManager()
		login_manager = lib.Networking.Login.NetworkLoginManager()

		engine = lib.Networking.NetworkEngine( mock_controller, bandwidth_manager, session_manager, domain_manager, login_manager )

		self.assertFalse( engine.IsRunning() )
		self.assertFalse( engine.IsShutdown() )

		mock_controller.CallToThread( engine.MainLoop )

		#

		with HTTMock( catch_all ):
			with HTTMock( catch_wew_ok ):
				job = lib.Networking.Jobs.NetworkJob( 'GET', MOCK_URL )

				engine.AddJob( job )

				time.sleep( 0.1 )

				self.assertTrue( job.IsDone() )
				self.assertFalse( job.HasError() )

				engine._new_work_to_do.set()

				time.sleep( 0.1 )

				self.assertEqual( len( engine._jobs_awaiting_validity ), 0 )
				self.assertEqual( len( engine._jobs_awaiting_bandwidth ), 0 )
				self.assertEqual( len( engine._jobs_awaiting_login ), 0 )
				self.assertEqual( len( engine._jobs_awaiting_slot ), 0 )
				self.assertEqual( len( engine._jobs_running ), 0 )

		#

		engine.Shutdown()

class TestNetworkingJob( unittest.TestCase ):
	def _GetJob( self, for_login = False ):
		job = lib.Networking.Jobs.NetworkJob( 'GET', MOCK_URL )

		job.SetForLogin( for_login )

		mock_controller = TestController.MockController()
		bandwidth_manager = lib.Networking.Bandwidth.NetworkBandwidthManager()
		session_manager = lib.Networking.Sessions.NetworkSessionManager()
		domain_manager = lib.Networking.Domain.NetworkDomainManager()
		login_manager = lib.Networking.Login.NetworkLoginManager()

		engine = lib.Networking.NetworkEngine( mock_controller, bandwidth_manager, session_manager, domain_manager, login_manager )

		job.engine = engine

		return job

	def test_cancelled_manually( self ):
		job = self._GetJob()

		self.assertFalse( job.IsCancelled() )
		self.assertFalse( job.IsDone() )

		job.Cancel()

		self.assertTrue( job.IsCancelled() )
		self.assertTrue( job.IsDone() )

	def test_cancelled_app_shutdown( self ):
		job = self._GetJob()

		self.assertFalse( job.IsCancelled() )
		self.assertFalse( job.IsDone() )

		job.engine.controller.model_is_shutdown = True

		self.assertTrue( job.IsCancelled() )
		self.assertTrue( job.IsDone() )

	def test_sleep( self ):
		job = self._GetJob()

		self.assertFalse( job.IsAsleep() )

		job.Sleep( 3 )

		self.assertTrue( job.IsAsleep() )

		five_secs_from_now = lib.Hydrus.Data.GetNow() + 5

		with patch.object( HydrusData, 'GetNow', return_value = five_secs_from_now ):
			self.assertFalse( job.IsAsleep() )

	def test_bandwidth_exceeded( self ):
		RESTRICTIVE_DATA_RULES = lib.Networking.BandwidthRules()

		RESTRICTIVE_DATA_RULES.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 10 )

		DOMAIN_NETWORK_CONTEXT = lib.Networking.Contexts.NetworkContext( CC.NETWORK_CONTEXT_DOMAIN, MOCK_DOMAIN )

		#

		job = self._GetJob()

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.ReportDataUsed( [ DOMAIN_NETWORK_CONTEXT ], 50 )

		job.engine.bandwidth_manager.SetRules( DOMAIN_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), False )

		#

		job = self._GetJob( for_login = True )

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.ReportDataUsed( [ DOMAIN_NETWORK_CONTEXT ], 50 )

		job.engine.bandwidth_manager.SetRules( DOMAIN_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), True )

	def test_bandwidth_ok( self ):
		PERMISSIVE_DATA_RULES = lib.Networking.BandwidthRules()

		PERMISSIVE_DATA_RULES.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 1048576 )

		DOMAIN_NETWORK_CONTEXT = lib.Networking.Contexts.NetworkContext( CC.NETWORK_CONTEXT_DOMAIN, MOCK_DOMAIN )

		#

		job = self._GetJob()

		job.engine.bandwidth_manager.ReportDataUsed( [ DOMAIN_NETWORK_CONTEXT ], 50 )

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.SetRules( DOMAIN_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), True )

		#

		job = self._GetJob( for_login = True )

		job.engine.bandwidth_manager.ReportDataUsed( [ DOMAIN_NETWORK_CONTEXT ], 50 )

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.SetRules( DOMAIN_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), True )

	def test_bandwidth_reported( self ):
		with HTTMock( catch_all ):
			with HTTMock( catch_wew_ok ):
				job = self._GetJob()

				job.BandwidthOK()

				job.Start()

				bm = job.engine.bandwidth_manager

				tracker = bm.GetTracker( lib.Networking.Contexts.GLOBAL_NETWORK_CONTEXT )

				self.assertTrue( tracker.GetUsage( HC.BANDWIDTH_TYPE_REQUESTS, None ), 1 )
				self.assertTrue( tracker.GetUsage( HC.BANDWIDTH_TYPE_DATA, None ), 256 )

	def test_done_ok( self ):
		with HTTMock( catch_all ):
			with HTTMock( catch_wew_ok ):
				job = self._GetJob()

				job.Start()

				self.assertFalse( job.HasError() )

				self.assertEqual( job.GetContentBytes(), GOOD_RESPONSE )

				self.assertEqual( job.GetStatus(), ( 'done!', 256, 256, None ) )

	def test_error( self ):
		with HTTMock( catch_all ):
			with HTTMock( catch_wew_error ):
				job = self._GetJob()

				job.Start()

				self.assertTrue( job.HasError() )

				self.assertEqual( job.GetContentBytes(), BAD_RESPONSE )

				self.assertEqual( type( job.GetErrorException() ), lib.Hydrus.Exceptions.ServerException )

				self.assertTrue( job.GetErrorText(), BAD_RESPONSE )

				self.assertEqual( job.GetStatus(), ( '500 - Internal Server Error', 18, 18, None ) )

	def test_generate_login_process( self ):
		# test the system works as expected

		pass

	def test_needs_login( self ):
		# test for both normal and login

		pass

class TestNetworkingJobHydrus( unittest.TestCase ):
	def _GetJob( self, for_login = False ):
		job = lib.Networking.Jobs.NetworkJobHydrus( MOCK_HYDRUS_SERVICE_KEY, 'GET', MOCK_HYDRUS_URL )

		job.SetForLogin( for_login )

		mock_controller = TestController.MockController()

		mock_service = lib.Hydrus.Services.Local.GenerateService( MOCK_HYDRUS_SERVICE_KEY, HC.TAG_REPOSITORY, 'test tag repo' )

		mock_services_manager = TestController.MockServicesManager( ( mock_service, ) )

		mock_controller.services_manager = mock_services_manager

		bandwidth_manager = lib.Networking.Bandwidth.NetworkBandwidthManager()
		session_manager = lib.Networking.Sessions.NetworkSessionManager()
		domain_manager = lib.Networking.Domain.NetworkDomainManager()
		login_manager = lib.Networking.Login.NetworkLoginManager()

		engine = lib.Networking.NetworkEngine( mock_controller, bandwidth_manager, session_manager, domain_manager, login_manager )

		job.engine = engine

		return job

	def test_bandwidth_exceeded( self ):
		RESTRICTIVE_DATA_RULES = lib.Networking.BandwidthRules()

		RESTRICTIVE_DATA_RULES.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 10 )

		HYDRUS_NETWORK_CONTEXT = lib.Networking.Contexts.NetworkContext( CC.NETWORK_CONTEXT_HYDRUS, MOCK_HYDRUS_SERVICE_KEY )

		#

		job = self._GetJob()

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.ReportDataUsed( [ HYDRUS_NETWORK_CONTEXT ], 50 )

		job.engine.bandwidth_manager.SetRules( HYDRUS_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), False )

		#

		job = self._GetJob( for_login = True )

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.ReportDataUsed( [ HYDRUS_NETWORK_CONTEXT ], 50 )

		job.engine.bandwidth_manager.SetRules( HYDRUS_NETWORK_CONTEXT, RESTRICTIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), True )

	def test_bandwidth_ok( self ):
		PERMISSIVE_DATA_RULES = lib.Networking.BandwidthRules()

		PERMISSIVE_DATA_RULES.AddRule( HC.BANDWIDTH_TYPE_DATA, None, 1048576 )

		HYDRUS_NETWORK_CONTEXT = lib.Networking.Contexts.NetworkContext( CC.NETWORK_CONTEXT_HYDRUS, MOCK_HYDRUS_SERVICE_KEY )

		#

		job = self._GetJob()

		job.engine.bandwidth_manager.ReportDataUsed( [ HYDRUS_NETWORK_CONTEXT ], 50 )

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.SetRules( HYDRUS_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), True )

		#

		job = self._GetJob( for_login = True )

		job.engine.bandwidth_manager.ReportDataUsed( [ HYDRUS_NETWORK_CONTEXT ], 50 )

		self.assertEqual( job.BandwidthOK(), True )

		job.engine.bandwidth_manager.SetRules( HYDRUS_NETWORK_CONTEXT, PERMISSIVE_DATA_RULES )

		self.assertEqual( job.BandwidthOK(), True )

	def test_bandwidth_reported( self ):
		pass

	def test_done_ok( self ):
		with HTTMock( catch_all ):
			with HTTMock( catch_hydrus_ok ):
				job = self._GetJob()

				job.Start()

				self.assertFalse( job.HasError() )

				self.assertEqual( job.GetContentBytes(), GOOD_RESPONSE )

				self.assertEqual( job.GetStatus(), ( 'done!', 256, 256, None ) )

	def test_error( self ):
		with HTTMock( catch_all ):
			with HTTMock( catch_hydrus_error ):
				job = self._GetJob()

				job.Start()

				self.assertTrue( job.HasError() )

				self.assertEqual( job.GetContentBytes(), BAD_RESPONSE )

				self.assertEqual( type( job.GetErrorException() ), lib.Hydrus.Exceptions.ServerException )

				self.assertTrue( job.GetErrorText(), BAD_RESPONSE )

				self.assertEqual( job.GetStatus(), ( '500 - Internal Server Error', 18, 18, None ) )

	def test_generate_login_process( self ):
		# test the system works as expected

		pass

	def test_needs_login( self ):
		# test for both normal and login

		pass


code:
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.Threading
import lib.Hydrus.Constants as HC
import lib.Hydrus.Exceptions
import lib.Hydrus.Globals as HG
import os
import threading
import time
import unittest

def do_read_job( rwlock, result_list, name ):
	with rwlock.read:
		result_list.append( 'begin read {}'.format( name ) )

		time.sleep( 0.25 )

		result_list.append( 'end read {}'.format( name ) )

def do_write_job( rwlock, result_list, name ):
	with rwlock.write:
		result_list.append( 'begin write {}'.format( name ) )

		time.sleep( 0.5 )

		result_list.append( 'end write {}'.format( name ) )

class TestFileRWLock( unittest.TestCase ):
	def test_simple( self ):
		rwlock = lib.Hydrus.Threading.FileRWLock()

		result_list = []

		do_read_job( rwlock, result_list, '1' )

		expected_result = []

		expected_result.extend( [ 'begin read 1', 'end read 1' ] )

		self.assertEqual( result_list, expected_result )

		#

		result_list = []

		do_write_job( rwlock, result_list, '1' )

		expected_result = []

		expected_result.extend( [ 'begin write 1', 'end write 1' ] )

		self.assertEqual( result_list, expected_result )

	def test_shared_read( self ):
		rwlock = lib.Hydrus.Threading.FileRWLock()

		result_list = []

		HG.test_controller.CallLater( 0.0, do_read_job, rwlock, result_list, '1' )
		HG.test_controller.CallLater( 0.1, do_read_job, rwlock, result_list, '2' )

		time.sleep( 0.2 )

		with rwlock.write:
			pass

		results = set( result_list )

		expected_results = set()

		expected_results.update( [ 'begin read 1', 'end read 1' ] )
		expected_results.update( [ 'begin read 2', 'end read 2' ] )

		self.assertEqual( results, expected_results )

		#

		result_list = []

		for i in range( 10 ):
			HG.test_controller.CallLater( 0.0, do_read_job, rwlock, result_list, str( i ) )

		time.sleep( 0.2 )

		with rwlock.write:
			pass

		expected_results = set()

		for i in range( 10 ):
			expected_results.update( [ 'begin read {}'.format( i ), 'end read {}'.format( i ) ] )

		results = set( result_list )

		self.assertEqual( results, expected_results )

	def test_competing_write( self ):
		rwlock = lib.Hydrus.Threading.FileRWLock()

		result_list = []

		HG.test_controller.CallLater( 0.0, do_write_job, rwlock, result_list, '1' )
		HG.test_controller.CallLater( 0.1, do_write_job, rwlock, result_list, '2' )

		time.sleep( 0.2 )

		with rwlock.read:
			pass

		expected_result = []

		expected_result.extend( [ 'begin write 1', 'end write 1' ] )
		expected_result.extend( [ 'begin write 2', 'end write 2' ] )

		self.assertEqual( result_list, expected_result )

		#

		result_list = []

		for i in range( 10 ):
			HG.test_controller.CallLater( 0.0, do_write_job, rwlock, result_list, str( i ) )

		time.sleep( 0.2 )

		with rwlock.read:
			pass

		expected_results = set()

		for i in range( 10 ):
			expected_pair = ( 'begin write {}'.format( i ), 'end write {}'.format( i ) )

			expected_results.add( expected_pair )

		for i in range( 10 ):
			result_pair = ( result_list[ i * 2 ], result_list[ ( i * 2 ) + 1 ] )

			self.assertIn( result_pair, expected_results )

	def test_mixed_competing( self ):
		rwlock = lib.Hydrus.Threading.FileRWLock()

		result_list = []

		all_expected_results = set()

		for i in range( 10 ):
			HG.test_controller.CallLater( 0.0 * i, do_read_job, rwlock, result_list, str( i ) )

			all_expected_results.update( [ 'begin read {}'.format( i ), 'end read {}'.format( i ) ] )

		for i in range( 5 ):
			HG.test_controller.CallLater( 0.0 * i, do_write_job, rwlock, result_list, str( i ) )

			all_expected_results.update( [ 'begin write {}'.format( i ), 'end write {}'.format( i ) ] )

		time.sleep( 0.2 )

		with rwlock.read:
			pass

		with rwlock.write:
			pass

		self.assertEqual( set( result_list ), all_expected_results )

		for i in range( 15 ):
			# even if reads mix up a bit, every two items should have the same read/write value

			a = result_list[ i * 2 ]
			b = result_list[ ( i * 2 ) + 1 ]

			if 'read' in a:
				self.assertIn( 'read', b )

			if 'write' in a:
				self.assertTrue( a.startswith( 'begin write' ) )
				self.assertTrue( b.startswith( 'end write' ) )

				self.assertEqual( a[-2:], b[-2:] )

license: wtfpl


code:
import collections
import os
import random
import threading
import collections
import shutil
import sys
import tempfile
import time
import traceback
import unittest
import wx
import lib.Hydrus.Constants as HC
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.Globals as HG
import lib.Hydrus.API.Local
import lib.Hydrus.Defaults
import lib.Networking
import lib.Networking.Bandwidth
import lib.Networking.Domain
import lib.Networking.Login
import lib.Networking.Sessions
import lib.Hydrus.Services.Local
import lib.Hydrus.Threading
import lib.Hydrus.Exceptions
import lib.Networking.PubSub
import lib.Hydrus.Sessions
import lib.Hydrus.Tags
import lib.Hydrus.Threading
from . import TestClientAPI
from . import TestClientConstants
from . import TestClientDaemons
from . import TestClientData
from . import TestClientImageHandling
from . import TestClientImportOptions
from . import TestClientImportSubscriptions
from . import TestClientListBoxes
from . import TestClientNetworking
from . import TestClientThreading
from . import TestDialogs
from . import TestDB
from . import TestFunctions
from . import TestHydrusNATPunch
from . import TestHydrusNetworking
from . import TestSerialisableBase
from . import TestHydrusServer
from . import TestHydrusSessions
from . import TestHydrusTags
from twisted.internet import reactor
import lib.Hydrus.Caches
import lib.Hydrus.Data.Local
import lib.Hydrus.Options
import lib.Hydrus.Data
import lib.Hydrus.Paths

DB_DIR = None

tiniest_gif = b'\x47\x49\x46\x38\x39\x61\x01\x00\x01\x00\x00\xFF\x00\x2C\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x00\x3B'

LOCAL_RATING_LIKE_SERVICE_KEY = lib.Hydrus.Data.GenerateKey()
LOCAL_RATING_NUMERICAL_SERVICE_KEY = lib.Hydrus.Data.GenerateKey()

def ConvertServiceKeysToContentUpdatesToComparable( service_keys_to_content_updates ):
	comparable_dict = {}

	for ( service_key, content_updates ) in list( service_keys_to_content_updates.items() ):
		comparable_dict[ service_key ] = set( content_updates )

	return comparable_dict

class MockController( object ):
	def __init__( self ):

		self.model_is_shutdown = False

		self.new_options = lib.Hydrus.Options.ClientOptions()

	def CallToThread( self, callable, *args, **kwargs ):
		return HG.test_controller.CallToThread( callable, *args, **kwargs )

	def JustWokeFromSleep( self ):
		return False

	def ModelIsShutdown( self ):
		return self.model_is_shutdown or HG.test_controller.ModelIsShutdown()

	def pub( self, *args, **kwargs ):
		pass

	def sub( self, *args, **kwargs ):
		pass

class MockServicesManager( object ):
	def __init__( self, services ):
		self._service_keys_to_services = { service.GetServiceKey() : service for service in services }

	def GetName( self, service_key ):
		return self._service_keys_to_services[ service_key ].GetName()

	def GetService( self, service_key ):
		return self._service_keys_to_services[ service_key ]

	def ServiceExists( self, service_key ):
		return service_key in self._service_keys_to_services

class FakeWebSessionManager():
	def EnsureLoggedIn( self, name ):
		pass

	def GetCookies( self, *args, **kwargs ):
		return { 'session_cookie' : 'blah' }

class TestFrame( wx.Frame ):
	def __init__( self ):
		wx.Frame.__init__( self, None )

	def SetPanel( self, panel ):
		vbox = wx.BoxSizer( wx.VERTICAL )

		vbox.Add( panel, CC.FLAGS_EXPAND_BOTH_WAYS )

		self.SetSizer( vbox )

		self.Fit()

		self.Show()

only_run = None

class Controller( object ):
	def __init__( self, win, only_run ):
		self.win = win
		self.only_run = only_run

		self.db_dir = tempfile.mkdtemp()

		global DB_DIR

		DB_DIR = self.db_dir

		self._server_files_dir = os.path.join( self.db_dir, 'server_files' )
		self._updates_dir = os.path.join( self.db_dir, 'test_updates' )

		client_files_default = os.path.join( self.db_dir, 'client_files' )

		lib.Hydrus.Paths.MakeSureDirectoryExists( self._server_files_dir )
		lib.Hydrus.Paths.MakeSureDirectoryExists( self._updates_dir )
		lib.Hydrus.Paths.MakeSureDirectoryExists( client_files_default )

		HG.controller = self
		HG.client_controller = self
		HG.server_controller = self
		HG.test_controller = self

		self.gui = self

		self._call_to_threads = []

		self._pubsub = lib.Networking.PubSub.HydrusPubSub( self )

		self.new_options = lib.Hydrus.Options.ClientOptions()

		HC.options = lib.Hydrus.Defaults.GetClientDefaultOptions()

		self.options = HC.options

		def show_text( text ): pass

		lib.Hydrus.Data.ShowText = show_text

		self._reads = {}

		self._reads[ 'local_booru_share_keys' ] = []
		self._reads[ 'messaging_sessions' ] = []
		self._reads[ 'tag_censorship' ] = []
		self._reads[ 'options' ] = lib.Hydrus.Defaults.GetClientDefaultOptions()
		self._reads[ 'file_system_predicates' ] = []
		self._reads[ 'media_results' ] = []

		self.example_tag_repo_service_key = lib.Hydrus.Data.GenerateKey()

		services = []

		services.append( lib.Hydrus.Services.Local.GenerateService( CC.LOCAL_BOORU_SERVICE_KEY, HC.LOCAL_BOORU, 'local booru' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( CC.CLIENT_API_SERVICE_KEY, HC.CLIENT_API_SERVICE, 'client api' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( CC.COMBINED_LOCAL_FILE_SERVICE_KEY, HC.COMBINED_LOCAL_FILE, 'all local files' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( CC.LOCAL_FILE_SERVICE_KEY, HC.LOCAL_FILE_DOMAIN, 'my files' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( CC.TRASH_SERVICE_KEY, HC.LOCAL_FILE_TRASH_DOMAIN, 'trash' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( CC.LOCAL_TAG_SERVICE_KEY, HC.LOCAL_TAG, 'local tags' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( self.example_tag_repo_service_key, HC.TAG_REPOSITORY, 'example tag repo' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( CC.COMBINED_TAG_SERVICE_KEY, HC.COMBINED_TAG, 'all known tags' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( LOCAL_RATING_LIKE_SERVICE_KEY, HC.LOCAL_RATING_LIKE, 'example local rating like service' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( LOCAL_RATING_NUMERICAL_SERVICE_KEY, HC.LOCAL_RATING_NUMERICAL, 'example local rating numerical service' ) )

		self._reads[ 'services' ] = services

		client_files_locations = {}

		for prefix in lib.Hydrus.Data.IterateHexPrefixes():
			for c in ( 'f', 't' ):
				client_files_locations[ c + prefix ] = client_files_default

		self._reads[ 'client_files_locations' ] = client_files_locations

		self._reads[ 'sessions' ] = []
		self._reads[ 'tag_parents' ] = {}
		self._reads[ 'tag_siblings' ] = {}
		self._reads[ 'in_inbox' ] = False

		self._writes = collections.defaultdict( list )

		self._managers = {}

		self.services_manager = lib.Hydrus.Caches.ServicesManager( self )
		self.client_files_manager = lib.Hydrus.Caches.ClientFilesManager( self )

		self.parsing_cache = lib.Hydrus.Caches.ParsingCache()

		bandwidth_manager = lib.Networking.Bandwidth.NetworkBandwidthManager()
		session_manager = lib.Networking.Sessions.NetworkSessionManager()
		domain_manager = lib.Networking.Domain.NetworkDomainManager()

		lib.Hydrus.Defaults.SetDefaultDomainManagerData( domain_manager )

		login_manager = lib.Networking.Login.NetworkLoginManager()

		self.network_engine = lib.Networking.NetworkEngine( self, bandwidth_manager, session_manager, domain_manager, login_manager )

		self.CallToThreadLongRunning( self.network_engine.MainLoop )

		self.tag_censorship_manager = lib.Hydrus.Caches.TagCensorshipManager( self )
		self.tag_siblings_manager = lib.Hydrus.Caches.TagSiblingsManager( self )
		self.tag_parents_manager = lib.Hydrus.Caches.TagParentsManager( self )
		self._managers[ 'undo' ] = lib.Hydrus.Caches.UndoManager( self )
		self.server_session_manager = lib.Hydrus.Sessions.HydrusSessionManagerServer()

		self.bitmap_manager = lib.Hydrus.Caches.BitmapManager( self )

		self.local_booru_manager = lib.Hydrus.Caches.LocalBooruCache( self )
		self.client_api_manager = lib.Hydrus.API.Local.APIManager()

		self._cookies = {}

		self._job_scheduler = lib.Hydrus.Threading.JobScheduler( self )

		self._job_scheduler.start()

	def _GetCallToThread( self ):
		for call_to_thread in self._call_to_threads:
			if not call_to_thread.CurrentlyWorking():
				return call_to_thread

		if len( self._call_to_threads ) > 100:
			raise Exception( 'Too many call to threads!' )

		call_to_thread = lib.Hydrus.Threading.THREADCallToThread( self, 'CallToThread' )

		self._call_to_threads.append( call_to_thread )

		call_to_thread.start()

		return call_to_thread

	def _SetupWx( self ):
		self.locale = wx.Locale( wx.LANGUAGE_DEFAULT ) # Very important to init this here and keep it non garbage collected

		CC.GlobalBMPs.STATICInitialise()

		self.frame_icon = wx.Icon( os.path.join( HC.STATIC_DIR, 'hydrus_32_non-transparent.png' ), wx.BITMAP_TYPE_PNG )

	def pub( self, topic, *args, **kwargs ):
		pass

	def pubimmediate( self, topic, *args, **kwargs ):
		self._pubsub.pubimmediate( topic, *args, **kwargs )

	def sub( self, object, method_name, topic ):
		self._pubsub.sub( object, method_name, topic )

	def AcquirePageKey( self ):
		return lib.Hydrus.Data.GenerateKey()

	def CallBlockingToWX( self, win, func, *args, **kwargs ):
		def wx_code( win, job_key ):
			try:
				if win is not None and not win:
					raise lib.Hydrus.Exceptions.WXDeadWindowException( 'Parent Window was destroyed before wx command was called!' )

				result = func( *args, **kwargs )

				job_key.SetVariable( 'result', result )

			except ( lib.Hydrus.Exceptions.WXDeadWindowException, lib.Hydrus.Exceptions.InsufficientCredentialsException, lib.Hydrus.Exceptions.ShutdownException ) as e:
				job_key.SetVariable( 'error', e )

			except Exception as e:
				job_key.SetVariable( 'error', e )

				lib.Hydrus.Data.Print( 'CallBlockingToWX just caught this error:' )
				lib.Hydrus.Data.DebugPrint( traceback.format_exc() )

			finally:
				job_key.Finish()

		job_key = lib.Hydrus.Threading.JobKey()

		job_key.Begin()

		wx.CallAfter( wx_code, win, job_key )

		while not job_key.IsDone():
			if HG.model_shutdown:
				raise lib.Hydrus.Exceptions.ShutdownException( 'Application is shutting down!' )

			time.sleep( 0.05 )

		if job_key.HasVariable( 'result' ):
			# result can be None, for wx_code that has no return variable

			result = job_key.GetIfHasVariable( 'result' )

			return result

		error = job_key.GetIfHasVariable( 'error' )

		if error is not None:
			raise error

		raise lib.Hydrus.Exceptions.ShutdownException()

	def CallToThread( self, callable, *args, **kwargs ):
		call_to_thread = self._GetCallToThread()

		call_to_thread.put( callable, *args, **kwargs )

	CallToThreadLongRunning = CallToThread

	def CallLater( self, initial_delay, func, *args, **kwargs ):
		call = lib.Hydrus.Data.Call( func, *args, **kwargs )

		job = lib.Hydrus.Threading.SchedulableJob( self, self._job_scheduler, initial_delay, call )

		self._job_scheduler.AddJob( job )

		return job

	def CallLaterWXSafe( self, window, initial_delay, func, *args, **kwargs ):
		call = lib.Hydrus.Data.Call( func, *args, **kwargs )

		job = lib.Hydrus.Threading.WXAwareJob( self, self._job_scheduler, window, initial_delay, call )

		self._job_scheduler.AddJob( job )

		return job

	def CallRepeating( self, initial_delay, period, func, *args, **kwargs ):
		call = lib.Hydrus.Data.Call( func, *args, **kwargs )

		job = lib.Hydrus.Threading.RepeatingJob( self, self._job_scheduler, initial_delay, period, call )

		self._job_scheduler.AddJob( job )

		return job

	def CallRepeatingWXSafe( self, window, initial_delay, period, func, *args, **kwargs ):
		call = lib.Hydrus.Data.Call( func, *args, **kwargs )

		job = lib.Hydrus.Threading.WXAwareRepeatingJob( self, self._job_scheduler, window, initial_delay, period, call )

		self._job_scheduler.AddJob( job )

		return job

	def ClearWrites( self, name ):
		if name in self._writes:
			del self._writes[ name ]

	def DBCurrentlyDoingJob( self ):
		return False

	def GetFilesDir( self ):
		return self._server_files_dir

	def GetNewOptions( self ):
		return self.new_options

	def GetManager( self, manager_type ):
		return self._managers[ manager_type ]

	def GetWrite( self, name ):
		write = self._writes[ name ]

		del self._writes[ name ]

		return write

	def ImportURLFromAPI( self, url, service_keys_to_tags, destination_page_name, show_destination_page ):
		normalised_url = self.network_engine.domain_manager.NormaliseURL( url )

		human_result_text = '"{}" URL added successfully.'.format( normalised_url )

		self.Write( 'import_url_test', url, service_keys_to_tags, destination_page_name, show_destination_page )

		return ( normalised_url, human_result_text )

	def IsBooted( self ):
		return True

	def IsCurrentPage( self, page_key ):
		return False

	def IsFirstStart( self ):
		return True

	def IShouldRegularlyUpdate( self, window ):
		return True

	def JustWokeFromSleep( self ):
		return False

	def ModelIsShutdown( self ):
		return HG.model_shutdown

	def PageAlive( self, page_key ):
		return False

	def PageClosedButNotDestroyed( self, page_key ):
		return False

	def Read( self, name, *args, **kwargs ):
		return self._reads[ name ]

	def RegisterUIUpdateWindow( self, window ):
		pass

	def ReleasePageKey( self, page_key ):
		pass

	def ReportDataUsed( self, num_bytes ):
		pass

	def ReportRequestUsed( self ):
		pass

	def ResetIdleTimer( self ): pass

	def Run( self, window ):
		# we are in wx thread here, we can do this
		self._SetupWx()

		suites = []

		if self.only_run is None:
			run_all = True

		else:
			run_all = False

		# the gui stuff runs fine on its own but crashes in the full test if it is not early, wew
		# something to do with the delayed button clicking stuff
		if run_all or self.only_run == 'gui':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestDialogs ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientListBoxes ) )

		if run_all or self.only_run == 'client_api':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientAPI ) )

		if run_all or self.only_run == 'daemons':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientDaemons ) )

		if run_all or self.only_run == 'data':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientConstants ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientData ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientImportOptions ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientThreading ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestFunctions ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestSerialisableBase ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestHydrusSessions ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestHydrusTags ) )

		if run_all or self.only_run == 'db':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestDB ) )

		if run_all or self.only_run == 'networking':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientNetworking ) )
			suites.append( unittest.TestLoader().loadTestsFromModule( TestHydrusNetworking ) )

		if run_all or self.only_run == 'import':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientImportSubscriptions ) )

		if run_all or self.only_run == 'image':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestClientImageHandling ) )

		if run_all or self.only_run == 'nat':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestHydrusNATPunch ) )

		if run_all or self.only_run == 'server':
			suites.append( unittest.TestLoader().loadTestsFromModule( TestHydrusServer ) )

		suite = unittest.TestSuite( suites )

		runner = unittest.TextTestRunner( verbosity = 2 )

		def do_it():
			try:
				runner.run( suite )

			finally:
				wx.CallAfter( self.win.DestroyLater )

		self.win.Show()

		test_thread = threading.Thread( target = do_it )

		test_thread.start()

	def SetRead( self, name, value ):
		self._reads[ name ] = value

	def SetStatusBarDirty( self ):
		pass

	def SetWebCookies( self, name, value ):
		self._cookies[ name ] = value

	def TidyUp( self ):
		time.sleep( 2 )

		lib.Hydrus.Paths.DeletePath( self.db_dir )

	def ViewIsShutdown( self ):
		return HG.view_shutdown

	def WaitUntilModelFree( self ):
		return

	def WaitUntilViewFree( self ):
		return

	def Write( self, name, *args, **kwargs ):
		self._writes[ name ].append( ( args, kwargs ) )

	def WriteSynchronous( self, name, *args, **kwargs ):
		self._writes[ name ].append( ( args, kwargs ) )

		if name == 'import_file':
			( file_import_job, ) = args

			if file_import_job.GetHash().hex() == 'a593942cb7ea9ffcd8ccf2f0fa23c338e23bfecd9a3e508dfc0bcf07501ead08': # 'blarg' in sha256 hex

				raise Exception( 'File failed to import for some reason!' )

			else:
				return ( CC.STATUS_SUCCESSFUL_AND_NEW, 'test note' )


code:
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.Data.Local
import lib.Hydrus.DB.Local
import lib.Hydrus.Defaults
import lib.Networking.Downloading
import lib.Actions.Export
import lib.Hydrus.Files.Local
import lib.GUI.Management
import lib.GUIPages
import lib.Actions.Import
import lib.Actions.Import.Local
import lib.Actions.Import.Options
import lib.Actions.Import.FileSeeds
import lib.Hydrus.Ratings
import lib.Hydrus.Search
import lib.Hydrus.Services.Local
import lib.Hydrus.Tags.Local
import collections
import lib.Hydrus.Constants as HC
import lib.Hydrus.Data
import lib.Hydrus.Exceptions
import lib.Handling.Video
import lib.Hydrus.Globals as HG
import lib.Networking
import lib.Hydrus.Serialisable
import itertools
import os
import lib.Hydrus.DB.Remote
import shutil
import sqlite3
import stat
from . import TestController
import time
import threading
import unittest
import wx

class TestClientDB( unittest.TestCase ):
	@classmethod

	def _clear_db( cls ):
		cls._delete_db()

		# class variable
		cls._db = lib.Hydrus.DB.Local.DB( HG.test_controller, TestController.DB_DIR, 'client' )

	@classmethod

	def _delete_db( cls ):
		cls._db.Shutdown()

		while not cls._db.LoopIsFinished():
			time.sleep( 0.1 )

		db_filenames = list( cls._db._db_filenames.values() )

		for filename in db_filenames:
			path = os.path.join( TestController.DB_DIR, filename )

			os.remove( path )

		del cls._db

	@classmethod

	def setUpClass( cls ):
		cls._db = lib.Hydrus.DB.Local.DB( HG.test_controller, TestController.DB_DIR, 'client' )

		HG.test_controller.SetRead( 'hash_status', ( CC.STATUS_UNKNOWN, None, '' ) )

	@classmethod

	def tearDownClass( cls ):
		cls._delete_db()

	def _read( self, action, *args, **kwargs ): return Testlib.Hydrus.DB.Local._db.Read( action, *args, **kwargs )
	def _write( self, action, *args, **kwargs ): return Testlib.Hydrus.DB.Local._db.Write( action, True, *args, **kwargs )

	def test_autocomplete( self ):
		Testlib.Hydrus.DB.Local._clear_db()

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'c*' )

		self.assertEqual( result, [] )

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'series:*' )

		self.assertEqual( result, [] )

		#

		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'

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

		file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

		file_import_job.GenerateHashAndStatus()

		file_import_job.GenerateInfo()

		self._write( 'import_file', file_import_job )

		#

		service_keys_to_content_updates = {}

		content_updates = []

		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'car', ( hash, ) ) ) )
		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'series:cars', ( hash, ) ) ) )
		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'maker:ford', ( hash, ) ) ) )

		service_keys_to_content_updates[ CC.LOCAL_TAG_SERVICE_KEY ] = content_updates

		self._write( 'content_updates', service_keys_to_content_updates )

		# cars

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'c*', add_namespaceless = True )

		preds = set()

		preds.add( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'car', min_current_count = 1 ) )
		preds.add( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'series:cars', min_current_count = 1 ) )

		for p in result:
			self.assertEqual( p.GetCount( HC.CONTENT_STATUS_CURRENT ), 1 )

		self.assertEqual( set( result ), preds )

		# cars

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'c*', add_namespaceless = False )

		preds = set()

		preds.add( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'series:cars', min_current_count = 1 ) )
		preds.add( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'car', min_current_count = 1 ) )

		for p in result:
			self.assertEqual( p.GetCount( HC.CONTENT_STATUS_CURRENT ), 1 )

		self.assertEqual( set( result ), preds )

		#

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'ser*' )

		preds = set()

		preds.add( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'series:cars', min_current_count = 1 ) )

		self.assertEqual( set( result ), preds )

		#

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'series:c*' )

		pred = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'series:cars', min_current_count = 1 )

		( read_pred, ) = result

		self.assertEqual( read_pred.GetCount( HC.CONTENT_STATUS_CURRENT ), 1 )

		self.assertEqual( pred, read_pred )

		#

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'car', exact_match = True )

		pred = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'car', min_current_count = 1 )

		( read_pred, ) = result

		self.assertEqual( read_pred.GetCount( HC.CONTENT_STATUS_CURRENT ), 1 )

		self.assertEqual( pred, read_pred )

		#

		result = self._read( 'autocomplete_predicates', tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, search_text = 'c', exact_match = True )

		self.assertEqual( result, [] )

	def test_export_folders( self ):
		file_search_context = lib.Hydrus.Search.FileSearchContext( file_service_key = lib.Hydrus.Data.GenerateKey(), tag_service_key = lib.Hydrus.Data.GenerateKey(), predicates = [ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'test' ) ] )

		export_folder = lib.Actions.Export.ExportFolder( 'test path', export_type = HC.EXPORT_FOLDER_TYPE_REGULAR, delete_from_client_after_export = False, file_search_context = file_search_context, period = 3600, phrase = '{hash}' )

		self._write( 'serialisable', export_folder )

		[ result ] = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["EXPORT_FOLDER"] )

		self.assertEqual( result.GetName(), export_folder.GetName() )

	def test_file_query_ids( self ):
		Testlib.Hydrus.DB.Local._clear_db()

		def run_namespace_predicate_tests( tests ):
			for ( inclusive, namespace, result ) in tests:
				predicates = [ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_NAMESPACE, namespace, inclusive ) ]

				search_context = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = predicates )

				file_query_ids = self._read( 'file_query_ids', search_context )

				self.assertEqual( len( file_query_ids ), result )

		def run_system_predicate_tests( tests ):
			for ( predicate_type, info, result ) in tests:
				predicates = [ lib.Hydrus.Search.Predicate( predicate_type, info ) ]

				search_context = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = predicates )

				file_query_ids = self._read( 'file_query_ids', search_context )

				self.assertEqual( len( file_query_ids ), result )

		def run_tag_predicate_tests( tests ):
			for ( inclusive, tag, result ) in tests:
				predicates = [ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, tag, inclusive ) ]

				search_context = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = predicates )

				file_query_ids = self._read( 'file_query_ids', search_context )

				self.assertEqual( len( file_query_ids ), result )

		def run_or_predicate_tests( tests ):
			for ( predicates, result ) in tests:
				search_context = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = predicates )

				file_query_ids = self._read( 'file_query_ids', search_context )

				self.assertEqual( len( file_query_ids ), result )

		tests = []

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_ARCHIVE, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_EVERYTHING, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_INBOX, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_LOCAL, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NOT_LOCAL, None, 0 ) )

		run_system_predicate_tests( tests )

		#

		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'

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

		file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

		file_import_job.GenerateHashAndStatus()

		file_import_job.GenerateInfo()

		( written_status, written_note ) = self._write( 'import_file', file_import_job )

		self.assertEqual( written_status, CC.STATUS_SUCCESSFUL_AND_NEW )
		self.assertEqual( written_note, '' )
		self.assertEqual( file_import_job.GetHash(), hash )

		time.sleep( 1 )

		#

		tests = []

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_AGE, ( '<', 'delta', ( 1, 1, 1, 1, ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_AGE, ( '<', 'delta', ( 0, 0, 0, 0, ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_AGE, ( '\u2248', 'delta', ( 1, 1, 1, 1, ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_AGE, ( '\u2248', 'delta', ( 0, 0, 0, 0, ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_AGE, ( '>', 'delta', ( 1, 1, 1, 1, ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_AGE, ( '>', 'delta', ( 0, 0, 0, 0, ) ), 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_ARCHIVE, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '<', 100, ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '<', 0, ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '\u2248', 100, ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '\u2248', 0, ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '=', 100, ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '=', 0, ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '>', 100, ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_DURATION, ( '>', 0, ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_EVERYTHING, None, 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_FILE_SERVICE, ( False, HC.CONTENT_STATUS_CURRENT, CC.LOCAL_FILE_SERVICE_KEY ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_FILE_SERVICE, ( False, HC.CONTENT_STATUS_PENDING, CC.LOCAL_FILE_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_FILE_SERVICE, ( True, HC.CONTENT_STATUS_CURRENT, CC.LOCAL_FILE_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_FILE_SERVICE, ( True, HC.CONTENT_STATUS_PENDING, CC.LOCAL_FILE_SERVICE_KEY ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HASH, ( hash, 'sha256' ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HASH, ( bytes.fromhex( '0123456789abcdef' * 4 ), 'sha256' ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '<', 201 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '<', 200 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '<', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '\u2248', 200 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '\u2248', 60 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '\u2248', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '=', 200 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '=', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '>', 200 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '>', 199 ), 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_INBOX, None, 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_LOCAL, None, 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_MIME, HC.IMAGES, 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_MIME, ( HC.IMAGE_PNG, ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_MIME, ( HC.IMAGE_JPEG, ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_MIME, HC.VIDEO, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NOT_LOCAL, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '<', 1 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '<', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '=', 0 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '=', 1 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '>', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '>', 1 ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '<', 1 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '<', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '\u2248', 0 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '\u2248', 1 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '=', 0 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '=', 1 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '>', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, ( '>', 1 ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATIO, ( '=', 1, 1 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATIO, ( '=', 4, 3 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATIO, ( '\u2248', 1, 1 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATIO, ( '\u2248', 200, 201 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATIO, ( '\u2248', 4, 1 ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIMILAR_TO, ( hash, 5 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIMILAR_TO, ( bytes.fromhex( '0123456789abcdef' * 4 ), 5 ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '<', 0, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '<', 5270, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '<', 5271, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '=', 5270, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '=', 0, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '\u2248', 5270, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '\u2248', 0, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '>', 5270, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '>', 5269, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '>', 0, lib.Hydrus.Data.ConvertUnitToInt( 'B' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '>', 0, lib.Hydrus.Data.ConvertUnitToInt( 'KB' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '>', 0, lib.Hydrus.Data.ConvertUnitToInt( 'MB' ) ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_SIZE, ( '>', 0, lib.Hydrus.Data.ConvertUnitToInt( 'GB' ) ), 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '<', 201 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '<', 200 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '<', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '\u2248', 200 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '\u2248', 60 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '\u2248', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '=', 200 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '=', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '>', 200 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_WIDTH, ( '>', 199 ), 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_LIMIT, 100, 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_LIMIT, 1, 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_LIMIT, 0, 0 ) )

		run_system_predicate_tests( tests )

		#

		service_keys_to_content_updates = {}

		service_keys_to_content_updates[ CC.COMBINED_LOCAL_FILE_SERVICE_KEY ] = ( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_ARCHIVE, ( hash, ) ), )
		service_keys_to_content_updates[ CC.LOCAL_TAG_SERVICE_KEY ] = ( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'car', ( hash, ) ) ), )

		self._write( 'content_updates', service_keys_to_content_updates )

		#

		tests = []

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_ARCHIVE, None, 1 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_INBOX, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '<', 2 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '<', 1 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '<', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '=', 0 ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '=', 1 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '>', 0 ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, ( '>', 1 ), 0 ) )

		run_system_predicate_tests( tests )

		#

		tests = []

		tests.append( ( True, 'car', 1 ) )
		tests.append( ( False, 'car', 0 ) )
		tests.append( ( True, 'bus', 0 ) )
		tests.append( ( False, 'bus', 1 ) )

		run_tag_predicate_tests( tests )

		#

		tests = []

		tests.append( ( True, 'series', 0 ) )
		tests.append( ( False, 'series', 1 ) )

		run_namespace_predicate_tests( tests )

		#

		service_keys_to_content_updates = {}

		content_updates = []

		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'series:cars', ( hash, ) ) ) )
		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'maker:ford', ( hash, ) ) ) )

		service_keys_to_content_updates[ CC.LOCAL_TAG_SERVICE_KEY ] = content_updates

		self._write( 'content_updates', service_keys_to_content_updates )

		#

		tests = []

		tests.append( ( True, 'maker:ford', 1 ) )
		tests.append( ( True, 'ford', 1 ) )
		tests.append( ( False, 'maker:ford', 0 ) )
		tests.append( ( False, 'ford', 0 ) )

		run_tag_predicate_tests( tests )

		#

		tests = []

		tests.append( ( True, 'series', 1 ) )
		tests.append( ( False, 'series', 0 ) )

		run_namespace_predicate_tests( tests )

		#

		tests = []

		preds = []

		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'car' ) )
		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'bus' ) )

		or_pred = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_OR_CONTAINER, preds )

		tests.append( ( [ or_pred ], 1 ) )

		preds = []

		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'car' ) )
		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_SYSTEM_HEIGHT, ( '<', 201 ) ) )

		or_pred = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_OR_CONTAINER, preds )

		tests.append( ( [ or_pred ], 1 ) )

		preds = []

		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'truck' ) )
		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'bus' ) )

		or_pred = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_OR_CONTAINER, preds )

		tests.append( ( [ or_pred ], 0 ) )

		preds = []

		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'truck', inclusive = False ) )
		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'bus' ) )

		or_pred = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_OR_CONTAINER, preds )

		tests.append( ( [ or_pred ], 1 ) )

		preds = []

		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'car' ) )
		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'truck' ) )

		or_pred_1 = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_OR_CONTAINER, preds )

		preds = []

		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'maker:toyota' ) )
		preds.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'maker:ford' ) )

		or_pred_2 = lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_OR_CONTAINER, preds )

		tests.append( ( [ or_pred_1, or_pred_2 ], 1 ) )

		run_or_predicate_tests( tests )

		#

		from . import TestController

		services = self._read( 'services' )

		services.append( lib.Hydrus.Services.Local.GenerateService( TestController.LOCAL_RATING_LIKE_SERVICE_KEY, HC.LOCAL_RATING_LIKE, 'test like rating service' ) )
		services.append( lib.Hydrus.Services.Local.GenerateService( TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY, HC.LOCAL_RATING_NUMERICAL, 'test numerical rating service' ) )

		self._write( 'update_services', services )

		service_keys_to_content_updates = {}

		content_updates = []

		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 1.0, ( hash, ) ) ) )

		service_keys_to_content_updates[ TestController.LOCAL_RATING_LIKE_SERVICE_KEY ] = content_updates

		self._write( 'content_updates', service_keys_to_content_updates )

		service_keys_to_content_updates = {}

		content_updates = []

		content_updates.append( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_RATINGS, HC.CONTENT_UPDATE_ADD, ( 0.6, ( hash, ) ) ) )

		service_keys_to_content_updates[ TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ] = content_updates

		self._write( 'content_updates', service_keys_to_content_updates )

		tests = []

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 1.0, TestController.LOCAL_RATING_LIKE_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 0.0, TestController.LOCAL_RATING_LIKE_SERVICE_KEY ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 'rated', TestController.LOCAL_RATING_LIKE_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 'not rated', TestController.LOCAL_RATING_LIKE_SERVICE_KEY ), 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 0.6, TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 1.0, TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '>', 0.6, TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ), 0 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '>', 0.4, TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 'rated', TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ), 1 ) )
		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '=', 'not rated', TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ), 0 ) )

		run_system_predicate_tests( tests )

		#

		content_update = lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_DELETE, ( hash, ), reason = 'test delete' )

		service_keys_to_content_updates = { CC.LOCAL_FILE_SERVICE_KEY : ( content_update, ) }

		self._write( 'content_updates', service_keys_to_content_updates )

		#

		tests = []

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_ARCHIVE, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_EVERYTHING, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_INBOX, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_LOCAL, None, 0 ) )

		tests.append( ( HC.PREDICATE_TYPE_SYSTEM_NOT_LOCAL, None, 0 ) )

		run_system_predicate_tests( tests )

	def test_file_system_predicates( self ):
		Testlib.Hydrus.DB.Local._clear_db()

		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'

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

		file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

		file_import_job.GenerateHashAndStatus()

		file_import_job.GenerateInfo()

		self._write( 'import_file', file_import_job )

		#

		result = self._read( 'file_system_predicates', CC.LOCAL_FILE_SERVICE_KEY )

		predicates = []

		predicates.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_SYSTEM_EVERYTHING, min_current_count = 1 ) )
		predicates.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_SYSTEM_INBOX, min_current_count = 1 ) )
		predicates.append( lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_SYSTEM_ARCHIVE, min_current_count = 0 ) )
		predicates.extend( [ lib.Hydrus.Search.Predicate( predicate_type ) for predicate_type in [ HC.PREDICATE_TYPE_SYSTEM_UNTAGGED, HC.PREDICATE_TYPE_SYSTEM_NUM_TAGS, HC.PREDICATE_TYPE_SYSTEM_LIMIT, HC.PREDICATE_TYPE_SYSTEM_SIZE, HC.PREDICATE_TYPE_SYSTEM_AGE, HC.PREDICATE_TYPE_SYSTEM_KNOWN_URLS, HC.PREDICATE_TYPE_SYSTEM_HASH, HC.PREDICATE_TYPE_SYSTEM_DIMENSIONS, HC.PREDICATE_TYPE_SYSTEM_DURATION, HC.PREDICATE_TYPE_SYSTEM_NUM_WORDS, HC.PREDICATE_TYPE_SYSTEM_MIME, HC.PREDICATE_TYPE_SYSTEM_SIMILAR_TO, HC.PREDICATE_TYPE_SYSTEM_FILE_SERVICE, HC.PREDICATE_TYPE_SYSTEM_TAG_AS_NUMBER, HC.PREDICATE_TYPE_SYSTEM_DUPLICATE_RELATIONSHIPS, HC.PREDICATE_TYPE_SYSTEM_FILE_VIEWING_STATS ] ] )

		self.assertEqual( set( result ), set( predicates ) )

		for i in range( len( predicates ) ):
			self.assertEqual( result[i].GetCount(), predicates[i].GetCount() )

	def test_gui_sessions( self ):
		def wx_code():
			test_frame = TestController.TestFrame()

			try:
				session = ClientGUIPages.GUISession( 'test_session' )

				#

				management_controller = lib.GUI.Management.CreateManagementControllerImportGallery()

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				management_controller = lib.GUI.Management.CreateManagementControllerImportMultipleWatcher()

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				service_keys_to_tags = lib.Hydrus.Tags.Local.ServiceKeysToTags( { lib.Hydrus.Data.GenerateKey() : [ 'some', 'tags' ] } )

				management_controller = lib.GUI.Management.CreateManagementControllerImportHDD( [ 'some', 'paths' ], lib.Actions.Import.Options.FileImportOptions(), { 'paths' : service_keys_to_tags }, True )

				management_controller.GetVariable( 'hdd_import' ).PausePlay() # to stop trying to import 'some' 'paths'

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				management_controller = lib.GUI.Management.CreateManagementControllerImportSimpleDownloader()

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				management_controller = lib.GUI.Management.CreateManagementControllerPetitions( HG.test_controller.example_tag_repo_service_key )

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				fsc = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = [] )

				management_controller = lib.GUI.Management.CreateManagementControllerQuery( 'search', CC.LOCAL_FILE_SERVICE_KEY, fsc, True )

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				fsc = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, tag_service_key = CC.LOCAL_TAG_SERVICE_KEY, predicates = [] )

				management_controller = lib.GUI.Management.CreateManagementControllerQuery( 'search', CC.LOCAL_FILE_SERVICE_KEY, fsc, False )

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [ lib.Hydrus.Data.GenerateKey() for i in range( 200 ) ] )

				session.AddPageTuple( page )

				#

				fsc = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = [ lib.Hydrus.Search.SYSTEM_PREDICATE_ARCHIVE ] )

				management_controller = lib.GUI.Management.CreateManagementControllerQuery( 'files', CC.LOCAL_FILE_SERVICE_KEY, fsc, True )

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				fsc = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = [ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_TAG, 'tag', min_current_count = 1, min_pending_count = 3 ) ] )

				management_controller = lib.GUI.Management.CreateManagementControllerQuery( 'wew lad', CC.LOCAL_FILE_SERVICE_KEY, fsc, True )

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				fsc = lib.Hydrus.Search.FileSearchContext( file_service_key = CC.LOCAL_FILE_SERVICE_KEY, predicates = [ lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_SYSTEM_RATING, ( '>', 0.2, TestController.LOCAL_RATING_NUMERICAL_SERVICE_KEY ) ), lib.Hydrus.Search.Predicate( HC.PREDICATE_TYPE_SYSTEM_FILE_SERVICE, ( True, HC.CONTENT_STATUS_CURRENT, CC.LOCAL_FILE_SERVICE_KEY ) ) ] )

				management_controller = lib.GUI.Management.CreateManagementControllerQuery( 'files', CC.LOCAL_FILE_SERVICE_KEY, fsc, True )

				page = ClientGUIPages.Page( test_frame, HG.test_controller, management_controller, [] )

				session.AddPageTuple( page )

				#

				self._write( 'serialisable', session )

				result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["GUI_SESSION"], 'test_session' )

				page_names = []

				for ( page_type, page_data ) in result.GetPageTuples():
					if page_type == 'page':
						( management_controller, initial_hashes ) = page_data

						page_names.append( management_controller.GetPageName() )

				self.assertEqual( page_names, [ 'gallery', 'watcher', 'import', 'simple downloader', 'example tag repo petitions', 'search', 'search', 'files', 'wew lad', 'files' ] )

			finally:
				test_frame.DestroyLater()

		HG.test_controller.CallBlockingToWX( HG.test_controller.win, wx_code )

	def test_import( self ):
		Testlib.Hydrus.DB.Local._clear_db()

		test_files = []

		test_files.append( ( 'muh_swf.swf', 'edfef9905fdecde38e0752a5b6ab7b6df887c3968d4246adc9cffc997e168cdf', 456774, HC.APPLICATION_FLASH, 400, 400, 33, 1, None ) )
		test_files.append( ( 'muh_mp4.mp4', '2fa293907144a046d043d74e9570b1c792cbfd77ee3f5c93b2b1a1cb3e4c7383', 570534, HC.VIDEO_MP4, 480, 480, 'mp4_duration', 151, None ) )
		test_files.append( ( 'muh_mpeg.mpeg', 'aebb10aaf3b27a5878fd2732ea28aaef7bbecef7449eaa759421c4ba4efff494', 772096, HC.VIDEO_MPEG, 720, 480, 3500, 105, None ) )
		test_files.append( ( 'muh_webm.webm', '55b6ce9d067326bf4b2fbe66b8f51f366bc6e5f776ba691b0351364383c43fcb', 84069, HC.VIDEO_WEBM, 640, 360, 4010, 120, None ) )
		test_files.append( ( 'muh_jpg.jpg', '5d884d84813beeebd59a35e474fa3e4742d0f2b6679faa7609b245ddbbd05444', 42296, HC.IMAGE_JPEG, 392, 498, None, None, None ) )
		test_files.append( ( 'muh_png.png', 'cdc67d3b377e6e1397ffa55edc5b50f6bdf4482c7a6102c6f27fa351429d6f49', 31452, HC.IMAGE_PNG, 191, 196, None, None, None ) )
		test_files.append( ( 'muh_apng.png', '9e7b8b5abc7cb11da32db05671ce926a2a2b701415d1b2cb77a28deea51010c3', 616956, HC.IMAGE_APNG, 500, 500, 'apng_duration', 47, None ) )
		test_files.append( ( 'muh_gif.gif', '00dd9e9611ebc929bfc78fde99a0c92800bbb09b9d18e0946cea94c099b211c2', 15660, HC.IMAGE_GIF, 329, 302, 600, 5, None ) )

		for ( filename, hex_hash, size, mime, width, height, duration, num_frames, num_words ) in test_files:
			path = os.path.join( HC.STATIC_DIR, 'testing', filename )

			hash = bytes.fromhex( hex_hash )

			file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

			file_import_job.GenerateHashAndStatus()

			file_import_job.GenerateInfo()

			( written_status, written_note ) = self._write( 'import_file', file_import_job )

			self.assertEqual( written_status, CC.STATUS_SUCCESSFUL_AND_NEW )
			self.assertEqual( written_note, '' )
			self.assertEqual( file_import_job.GetHash(), hash )

			file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

			file_import_job.GenerateHashAndStatus()

			file_import_job.GenerateInfo()

			( written_status, written_note ) = self._write( 'import_file', file_import_job )

			# would be redundant, but triggers the 'it is missing from db' hook
			self.assertEqual( written_status, CC.STATUS_SUCCESSFUL_AND_NEW )
			self.assertIn( 'already in the db', written_note )
			self.assertEqual( file_import_job.GetHash(), hash )

			written_hash = file_import_job.GetHash()

			( media_result, ) = self._read( 'media_results', ( written_hash, ) )

			( mr_file_info_manager, mr_tags_manager, mr_locations_manager, mr_ratings_manager ) = media_result.ToTuple()

			( mr_hash_id, mr_hash, mr_size, mr_mime, mr_width, mr_height, mr_duration, mr_num_frames, mr_num_words ) = mr_file_info_manager.ToTuple()

			mr_inbox = mr_locations_manager.GetInbox()

			now = lib.Hydrus.Data.GetNow()

			self.assertEqual( mr_hash, hash )
			self.assertEqual( mr_inbox, True )
			self.assertEqual( mr_size, size )
			self.assertEqual( mr_mime, mime )
			self.assertEqual( mr_width, width )
			self.assertEqual( mr_height, height )

			if duration == 'apng_duration': # diff ffmpeg versions report differently

				self.assertIn( mr_duration, ( 3133, 1880, 1800 ) )

			elif duration == 'mp4_duration':
				self.assertIn( mr_duration, ( 6266, 6290 ) )

			else:
				self.assertEqual( mr_duration, duration )

			self.assertEqual( mr_num_frames, num_frames )
			self.assertEqual( mr_num_words, num_words )

	def test_import_folders( self ):
		import_folder_1 = lib.Actions.Import.Local.ImportFolder( 'imp 1', path = TestController.DB_DIR, mimes = HC.VIDEO, publish_files_to_popup_button = False )
		import_folder_2 = lib.Actions.Import.Local.ImportFolder( 'imp 2', path = TestController.DB_DIR, mimes = HC.IMAGES, period = 1200, publish_files_to_popup_button = False )

		#

		result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["IMPORT_FOLDER"] )

		self.assertEqual( result, [] )

		#

		self._write( 'serialisable', import_folder_1 )
		self._write( 'serialisable', import_folder_2 )

		result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["IMPORT_FOLDER"] )

		for item in result:
			self.assertEqual( type( item ), lib.Actions.Import.Local.ImportFolder )

		#

		self._write( 'delete_serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["IMPORT_FOLDER"], 'imp 2' )

		result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["IMPORT_FOLDER"] )

		( item, ) = result

		self.assertEqual( item.GetName(), 'imp 1' )

	def test_init( self ):
		self.assertTrue( os.path.exists( TestController.DB_DIR ) )

		self.assertTrue( os.path.exists( os.path.join( TestController.DB_DIR, 'client.db' ) ) )

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

		self.assertTrue( os.path.exists( client_files_default ) )

		for prefix in lib.Hydrus.Data.IterateHexPrefixes():
			for c in ( 'f', 't' ):
				dir = os.path.join( client_files_default, c + prefix )

				self.assertTrue( os.path.exists( dir ) )

	def test_hash_status( self ):
		Testlib.Hydrus.DB.Local._clear_db()

		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'

		md5 = bytes.fromhex( 'fdadb2cae78f2dfeb629449cd005f2a2' )

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

		#

		result = self._read( 'hash_status', 'md5', md5 )

		self.assertEqual( result, ( CC.STATUS_UNKNOWN, None, '' ) )

		#

		file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

		file_import_job.GenerateHashAndStatus()

		file_import_job.GenerateInfo()

		self._write( 'import_file', file_import_job )

		#

		( status, written_hash, note ) = self._read( 'hash_status', 'md5', md5 )

		# would be redundant, but sometimes(?) triggers the 'it is missing from db' hook
		self.assertIn( status, ( CC.STATUS_UNKNOWN, CC.STATUS_SUCCESSFUL_BUT_REDUNDANT ) )
		self.assertEqual( written_hash, hash )
		if status == CC.STATUS_UNKNOWN:
			self.assertIn( 'already in the db', note )

		#

		content_update = lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_DELETE, ( hash, ), reason = 'test delete' )

		service_keys_to_content_updates = { CC.LOCAL_FILE_SERVICE_KEY : ( content_update, ) }

		self._write( 'content_updates', service_keys_to_content_updates )

		#

		( status, hash, note ) = self._read( 'hash_status', 'md5', md5 )

		self.assertEqual( ( status, hash ), ( CC.STATUS_DELETED, hash ) )

	def test_media_results( self ):
		Testlib.Hydrus.DB.Local._clear_db()

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

		file_import_job = lib.Actions.Import.FileSeeds.FileImportJob( path )

		file_import_job.GenerateHashAndStatus()

		file_import_job.GenerateInfo()

		self._write( 'import_file', file_import_job )

		hash = file_import_job.GetHash()

		#

		( media_result, ) = self._read( 'media_results', ( hash, ) )

		( mr_file_info_manager, mr_tags_manager, mr_locations_manager, mr_ratings_manager ) = media_result.ToTuple()

		( mr_hash_id, mr_hash, mr_size, mr_mime, mr_width, mr_height, mr_duration, mr_num_frames, mr_num_words ) = mr_file_info_manager.ToTuple()

		mr_inbox = mr_locations_manager.GetInbox()

		now = lib.Hydrus.Data.GetNow()

		self.assertEqual( mr_hash, hash )
		self.assertEqual( mr_inbox, True )
		self.assertEqual( mr_size, 5270 )
		self.assertEqual( mr_mime, HC.IMAGE_PNG )
		self.assertEqual( mr_hash, hash )
		self.assertEqual( mr_width, 200 )
		self.assertEqual( mr_height, 200 )
		self.assertEqual( mr_duration, None )
		self.assertEqual( mr_num_frames, None )
		self.assertEqual( mr_num_words, None )

		( media_result, ) = self._read( 'media_results_from_ids', ( 1, ) )

		( mr_file_info_manager, mr_tags_manager, mr_locations_manager, mr_ratings_manager ) = media_result.ToTuple()

		( mr_hash_id, mr_hash, mr_size, mr_mime, mr_width, mr_height, mr_duration, mr_num_frames, mr_num_words ) = mr_file_info_manager.ToTuple()

		mr_inbox = mr_locations_manager.GetInbox()

		now = lib.Hydrus.Data.GetNow()

		self.assertEqual( mr_hash, hash )
		self.assertEqual( mr_inbox, True )
		self.assertEqual( mr_size, 5270 )
		self.assertEqual( mr_mime, HC.IMAGE_PNG )
		self.assertEqual( mr_hash, hash )
		self.assertEqual( mr_width, 200 )
		self.assertEqual( mr_height, 200 )
		self.assertEqual( mr_duration, None )
		self.assertEqual( mr_num_frames, None )
		self.assertEqual( mr_num_words, None )

	def test_tag_censorship( self ):
		result = self._read( 'tag_censorship' )

		self.assertEqual( result, [] )

		result = self._read( 'tag_censorship', CC.LOCAL_TAG_SERVICE_KEY )

		self.assertEqual( result, ( True, [] ) )

		#

		info = []

		info.append( ( CC.LOCAL_TAG_SERVICE_KEY, False, [ ':', 'series:' ] ) )
		info.append( ( CC.LOCAL_FILE_SERVICE_KEY, True, [ ':' ] ) ) # bit dodgy, but whatever!

		self._write( 'tag_censorship', info )

		#

		result = self._read( 'tag_censorship' )

		self.assertEqual( len( result ), len( info ) )

		for row in result:
			self.assertIn( row, info )

		result = self._read( 'tag_censorship', CC.LOCAL_TAG_SERVICE_KEY )

		self.assertEqual( result, ( False, [ ':', 'series:' ] ) )

	def test_nums_pending( self ):
		result = self._read( 'nums_pending' )

		self.assertEqual( result, {} )

		# we can do more testing when I add repo service to this testing framework

	def test_pending( self ):
		service_key = lib.Hydrus.Data.GenerateKey()

		services = self._read( 'services' )

		old_services = list( services )

		services.append( lib.Hydrus.Services.Local.GenerateService( service_key, HC.TAG_REPOSITORY, 'new tag repo' ) )

		self._write( 'update_services', services )

		#

		hashes = [ os.urandom( 32 ) for i in range( 64 ) ]

		tags = [ 'this', 'is', 'a:test' ]

		content_updates = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_PEND, ( tag, hashes ) ) for tag in tags ]

		service_keys_to_content_updates = { service_key : content_updates }

		self._write( 'content_updates', service_keys_to_content_updates )

		result = self._read( 'pending', service_key )

		self.assertIsInstance( result, lib.Networking.ClientToServerUpdate )

		self.assertTrue( result.HasContent() )

		self.assertEqual( set( result.GetHashes() ), set( hashes ) )

		#

		self._write( 'update_services', old_services )

	def test_pixiv_account( self ):
		result = self._read( 'serialisable_simple', 'pixiv_account' )

		self.assertEqual( result, None )

		pixiv_id = 123456
		password = 'password'

		self._write( 'serialisable_simple', 'pixiv_account', ( pixiv_id, password ) )

		result = self._read( 'serialisable_simple', 'pixiv_account' )

		self.assertTrue( result, ( pixiv_id, password ) )

	def test_repo_downloads( self ):
		result = self._read( 'downloads' )

		self.assertEqual( result, set() )

		#

		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'

		service_keys_to_content_updates = {}

		service_keys_to_content_updates[ CC.COMBINED_LOCAL_FILE_SERVICE_KEY ] = ( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_PEND, ( hash, ) ), )

		self._write( 'content_updates', service_keys_to_content_updates )

		#

		result = self._read( 'downloads' )

		self.assertEqual( result, { hash } )

		#

		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'

		service_keys_to_content_updates = {}

		service_keys_to_content_updates[ CC.COMBINED_LOCAL_FILE_SERVICE_KEY ] = ( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_FILES, HC.CONTENT_UPDATE_RESCIND_PEND, ( hash, ) ), )

		self._write( 'content_updates', service_keys_to_content_updates )

		#

		result = self._read( 'downloads' )

		self.assertEqual( result, set() )

	def test_services( self ):
		result = self._read( 'services', ( HC.LOCAL_FILE_DOMAIN, HC.LOCAL_FILE_TRASH_DOMAIN, HC.COMBINED_LOCAL_FILE, HC.LOCAL_TAG ) )

		result_service_keys = { service.GetServiceKey() for service in result }

		self.assertEqual( { CC.TRASH_SERVICE_KEY, CC.LOCAL_FILE_SERVICE_KEY, CC.LOCAL_UPDATE_SERVICE_KEY, CC.COMBINED_LOCAL_FILE_SERVICE_KEY, CC.LOCAL_TAG_SERVICE_KEY }, result_service_keys )

		#

		result = self._read( 'service_info', CC.LOCAL_FILE_SERVICE_KEY )

		self.assertEqual( type( result ), dict )

		for ( k, v ) in list( result.items() ):
			self.assertEqual( type( k ), int )
			self.assertEqual( type( v ), int )

		#

		NUM_DEFAULT_SERVICES = 10

		services = self._read( 'services' )

		self.assertEqual( len( services ), NUM_DEFAULT_SERVICES )

		old_services = list( services )

		services.append( lib.Hydrus.Services.Local.GenerateService( lib.Hydrus.Data.GenerateKey(), HC.TAG_REPOSITORY, 'new service' ) )

		self._write( 'update_services', services )

		services = self._read( 'services' )

		self.assertEqual( len( services ), NUM_DEFAULT_SERVICES + 1 )

		self._write( 'update_services', old_services )

		services = self._read( 'services' )

		self.assertEqual( len( services ), NUM_DEFAULT_SERVICES )

	def test_shortcuts( self ):
		num_default = len( lib.Hydrus.Defaults.GetDefaultShortcuts() )

		result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["SHORTCUTS"] )

		self.assertEqual( len( result ), num_default )

		#

		for ( i, shortcuts ) in enumerate( lib.Hydrus.Defaults.GetDefaultShortcuts() ):
			name = 'shortcuts ' + str( i )

			shortcuts.SetName( name )

			self._write( 'serialisable', shortcuts )

			result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["SHORTCUTS"] )

			self.assertEqual( len( result ), num_default + 1 )

			result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["SHORTCUTS"], name )

			for ( shortcut, command ) in shortcuts:
				self.assertEqual( result.GetCommand( shortcut ).GetData(), command.GetData() )

			#

			self._write( 'delete_serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["SHORTCUTS"], name )

			result = self._read( 'serialisable_named', lib.Hydrus.Serialisable.HG.SerialisableType["SHORTCUTS"] )

			self.assertEqual( len( result ), num_default )

class TestServerDB( unittest.TestCase ):
	def _read( self, action, *args, **kwargs ): return Testlib.Hydrus.DB.Remote._db.Read( action, *args, **kwargs )
	def _write( self, action, *args, **kwargs ): return Testlib.Hydrus.DB.Remote._db.Write( action, True, *args, **kwargs )

	@classmethod

	def setUpClass( cls ):
		cls._db = lib.Hydrus.DB.Remote.DB( HG.test_controller, TestController.DB_DIR, 'server' )

	@classmethod

	def tearDownClass( cls ):
		cls._db.Shutdown()

		while not cls._db.LoopIsFinished():
			time.sleep( 0.1 )

		del cls._db

	def _test_account_creation( self ):
		result = self._read( 'account_types', self._tag_service_key )

		( service_admin_at, ) = result

		self.assertEqual( service_admin_at.GetTitle(), 'service admin' )
		self.assertEqual( service_admin_at.GetPermissions(), [ HC.GET_DATA, HC.POST_DATA, HC.POST_PETITIONS, HC.RESOLVE_PETITIONS, HC.MANAGE_USERS, HC.GENERAL_ADMIN ] )
		self.assertEqual( service_admin_at.GetMaxBytes(), None )
		self.assertEqual( service_admin_at.GetMaxRequests(), None )

		#

		user_at = lib.Hydrus.Data.AccountType( 'user', [ HC.GET_DATA, HC.POST_DATA ], ( 50000, 500 ) )

		edit_log = [ ( HC.ADD, user_at ) ]

		self._write( 'account_types', self._tag_service_key, edit_log )

		result = self._read( 'account_types', self._tag_service_key )

		( at_1, at_2 ) = result

		d = { at_1.GetTitle() : at_1, at_2.GetTitle() : at_2 }

		at = d[ 'user' ]

		self.assertEqual( at.GetPermissions(), [ HC.GET_DATA, HC.POST_DATA ] )
		self.assertEqual( at.GetMaxBytes(), 50000 )
		self.assertEqual( at.GetMaxRequests(), 500 )

		#

		user_at_diff = lib.Hydrus.Data.AccountType( 'user different', [ HC.GET_DATA ], ( 40000, None ) )

		edit_log = [ ( HC.EDIT, ( 'user', user_at_diff ) ) ]

		self._write( 'account_types', self._tag_service_key, edit_log )

		result = self._read( 'account_types', self._tag_service_key )

		( at_1, at_2 ) = result

		d = { at_1.GetTitle() : at_1, at_2.GetTitle() : at_2 }

		at = d[ 'user different' ]

		self.assertEqual( at.GetPermissions(), [ HC.GET_DATA ] )
		self.assertEqual( at.GetMaxBytes(), 40000 )
		self.assertEqual( at.GetMaxRequests(), None )

		#

		r_keys = self._read( 'registration_keys', self._tag_service_key, 5, 'user different', 86400 * 365 )

		self.assertEqual( len( r_keys ), 5 )

		for r_key in r_keys:
			self.assertEqual( len( r_key ), 32 )

		r_key = r_keys[0]

		access_key = self._read( 'access_key', self._tag_service_key, r_key )
		access_key_2 = self._read( 'access_key', self._tag_service_key, r_key )

		self.assertNotEqual( access_key, access_key_2 )

		self.assertRaises( lib.Hydrus.Exceptions.InsufficientCredentialsException, self._read, 'account_key_from_access_key', self._tag_service_key, access_key )

		account_key = self._read( 'account_key_from_access_key', self._tag_service_key, access_key_2 )

		self.assertRaises( lib.Hydrus.Exceptions.InsufficientCredentialsException, self._read, 'access_key', r_key )

	def _test_content_creation( self ):
		# create some tag and hashes business, try uploading a file, and test that

		# fetch content update, test it. I think that works

		pass

	def _test_init_server_admin( self ):
		result = self._read( 'access_key', HC.SERVER_ADMIN_KEY, b'init' )

		self.assertEqual( type( result ), bytes )
		self.assertEqual( len( result ), 32 )

		self._admin_access_key = result

		result = self._read( 'account_key_from_access_key', HC.SERVER_ADMIN_KEY, self._admin_access_key )

		self.assertEqual( type( result ), bytes )
		self.assertEqual( len( result ), 32 )

		self._admin_account_key = result

	def _test_service_creation( self ):
		self._tag_service_key = lib.Hydrus.Data.GenerateKey()
		self._file_service_key = lib.Hydrus.Data.GenerateKey()

		edit_log = []

		t_options = { 'max_monthly_data' : None, 'message' : 'tag repo message', 'port' : 100, 'upnp' : None }
		f_options = { 'max_monthly_data' : None, 'message' : 'file repo message', 'port' : 101, 'upnp' : None }

		edit_log.append( ( HC.ADD, ( self._tag_service_key, HC.TAG_REPOSITORY, t_options ) ) )
		edit_log.append( ( HC.ADD, ( self._file_service_key, HC.FILE_REPOSITORY, f_options ) ) )

		result = self._write( 'services', self._admin_account_key, edit_log )

		self.assertIn( self._tag_service_key, result )

		self._tag_service_admin_access_key = result[ self._tag_service_key ]

		self.assertEqual( type( self._tag_service_admin_access_key ), bytes )
		self.assertEqual( len( self._tag_service_admin_access_key ), 32 )

		self.assertIn( self._file_service_key, result )

		self._file_service_admin_access_key = result[ self._file_service_key ]

		self.assertEqual( type( self._tag_service_admin_access_key ), bytes )
		self.assertEqual( len( self._tag_service_admin_access_key ), 32 )

		#

		result = self._read( 'service_keys', HC.REPOSITORIES )

		self.assertEqual( set( result ), { self._tag_service_key, self._file_service_key } )

	def test_server( self ):
		self._test_init_server_admin()

		# broke since service rewrite
		#self._test_service_creation()

		#self._test_account_creation()

		#self._test_content_creation()


code:
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.Defaults
import lib.GUI.Dialogs
import lib.GUI.ScrolledPanels.Edit
import lib.GUI.ScrolledPanels.Management
import lib.GUI.TopLevelWindows
import lib.Hydrus.Threading
import collections
import lib.Hydrus.Constants as HC
import os
import unittest
import wx
import lib.Hydrus.Globals as HG

def HitButton( button ):
	wx.QueueEvent( button, wx.CommandEvent( commandEventType = wx.EVT_BUTTON.typeId, id = button.GetId() ) )

def HitCancelButton( window ):
	wx.QueueEvent( window, wx.CommandEvent( commandEventType = wx.EVT_BUTTON.typeId, id = wx.ID_CANCEL ) )

def HitOKButton( window ):
	wx.QueueEvent( window, wx.CommandEvent( commandEventType = wx.EVT_BUTTON.typeId, id = wx.ID_OK ) )

def CancelChildDialog( window ):
	children = window.GetChildren()

	for child in children:
		if isinstance( child, wx.Dialog ):
			HitCancelButton( child )

def OKChildDialog( window ):
	children = window.GetChildren()

	for child in children:
		if isinstance( child, wx.Dialog ):
			HitOKButton( child )

def PressKey( window, key ):
	window.SetFocus()

	uias = wx.UIActionSimulator()

	uias.Char( key )

class TestDBDialogs( unittest.TestCase ):
	def test_dialog_manage_subs( self ):
		def wx_code():
			title = 'subs test'

			with lib.GUI.TopLevelWindows.DialogEdit( None, title ) as dlg:
				panel = lib.GUI.ScrolledPanels.Edit.EditSubscriptionsPanel( dlg, [] )

				dlg.SetPanel( panel )

				HG.test_controller.CallLaterWXSafe( dlg, 2, panel.Add )

				HG.test_controller.CallLaterWXSafe( dlg, 4, OKChildDialog, panel )

				HG.test_controller.CallLaterWXSafe( dlg, 6, HitCancelButton, dlg )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_CANCEL )

		HG.test_controller.CallBlockingToWX( HG.test_controller.win, wx_code )

class TestNonDBDialogs( unittest.TestCase ):
	def test_dialog_choose_new_service_method( self ):
		def wx_code():
			with lib.GUI.Dialogs.DialogChooseNewServiceMethod( None ) as dlg:
				HitButton( dlg._register )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_OK )

				register = dlg.GetRegister()

				self.assertEqual( register, True )

			with lib.GUI.Dialogs.DialogChooseNewServiceMethod( None ) as dlg:
				HitButton( dlg._setup )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_OK )

				register = dlg.GetRegister()

				self.assertEqual( register, False )

			with lib.GUI.Dialogs.DialogChooseNewServiceMethod( None ) as dlg:
				HitCancelButton( dlg )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_CANCEL )

		HG.test_controller.CallBlockingToWX( HG.test_controller.win, wx_code )

	def test_dialog_finish_filtering( self ):
		def wx_code():
			with lib.GUI.Dialogs.DialogFinishFiltering( None, 'keep 3 files and delete 5 files?' ) as dlg:
				HitButton( dlg._back )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_CANCEL )

			with lib.GUI.Dialogs.DialogFinishFiltering( None, 'keep 3 files and delete 5 files?' ) as dlg:
				HitButton( dlg._commit )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_YES )

			with lib.GUI.Dialogs.DialogFinishFiltering( None, 'keep 3 files and delete 5 files?' ) as dlg:
				HitButton( dlg._forget )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_NO )

		HG.test_controller.CallBlockingToWX( HG.test_controller.win, wx_code )

	def test_dialog_yes_no( self ):
		def wx_code():
			with lib.GUI.Dialogs.DialogYesNo( None, 'hello' ) as dlg:
				HitButton( dlg._yes )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_YES )

			with lib.GUI.Dialogs.DialogYesNo( None, 'hello' ) as dlg:
				HitButton( dlg._no )

				result = dlg.ShowModal()

				self.assertEqual( result, wx.ID_NO )

		HG.test_controller.CallBlockingToWX( HG.test_controller.win, wx_code )


code:
import collections
import lib.Hydrus.Constants as HC
import lib.Hydrus.Data.Local
import lib.Hydrus.Tags.Local
import os
import unittest
import lib.Hydrus.Data
import lib.Hydrus.Constants.Local as CC

class TestFunctions( unittest.TestCase ):
	def test_dict_to_content_updates( self ):
		hash = lib.Hydrus.Data.GenerateKey()

		hashes = { hash }

		local_key = CC.LOCAL_TAG_SERVICE_KEY
		remote_key = lib.Hydrus.Data.GenerateKey()

		service_keys_to_tags = lib.Hydrus.Tags.Local.ServiceKeysToTags( { local_key : { 'a' } } )

		content_updates = { local_key : [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'a', hashes ) ) ] }

		self.assertEqual( lib.Hydrus.Data.Local.ConvertServiceKeysToTagsToServiceKeysToContentUpdates( { hash }, service_keys_to_tags ), content_updates )

		service_keys_to_tags = lib.Hydrus.Tags.Local.ServiceKeysToTags( { remote_key : { 'c' } } )

		content_updates = { remote_key : [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_PEND, ( 'c', hashes ) ) ] }

		self.assertEqual( lib.Hydrus.Data.Local.ConvertServiceKeysToTagsToServiceKeysToContentUpdates( { hash }, service_keys_to_tags ), content_updates )

		service_keys_to_tags = lib.Hydrus.Tags.Local.ServiceKeysToTags( { local_key : [ 'a', 'character:b' ], remote_key : [ 'c', 'series:d' ] } )

		content_updates = {}

		content_updates[ local_key ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'a', hashes ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_ADD, ( 'character:b', hashes ) ) ]
		content_updates[ remote_key ] = [ lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_PEND, ( 'c', hashes ) ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_PEND, ( 'series:d', hashes ) ) ]

		self.assertEqual( lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_PEND, 'c' ), lib.Hydrus.Data.ContentUpdate( HC.CONTENT_TYPE_MAPPINGS, HC.CONTENT_UPDATE_PEND, 'c' ) )
		self.assertEqual( lib.Hydrus.Data.Local.ConvertServiceKeysToTagsToServiceKeysToContentUpdates( { hash }, service_keys_to_tags ), content_updates )

	def test_number_conversion( self ):
		i = 123456789

		i_pretty = lib.Hydrus.Data.ToHumanInt( i )

		# this test only works on anglo computers; it is mostly so I can check it is working on mine

		self.assertEqual( i_pretty, '123,456,789' )


code:
import lib.Hydrus.Constants.Local as CC
import lib.Hydrus.Constants as HC
import lib.Networking.NATPunch
import os
import random
import time
import unittest

class TestNATPunch( unittest.TestCase ):
	def test_upnp( self ):
		internal_client = lib.Networking.NATPunch.GetLocalIP()

		internal_port = random.randint( 1000, 1500 )

		external_port = random.randint( 1000, 1500 )

		description_tcp = 'hydrus test tcp'
		description_udp = 'hydrus test udp'

		lib.Networking.NATPunch.AddUPnPMapping( internal_client, internal_port, external_port, 'TCP', description_tcp )
		lib.Networking.NATPunch.AddUPnPMapping( internal_client, internal_port, external_port, 'UDP', description_udp )

		mappings = lib.Networking.NATPunch.GetUPnPMappings()

		external_ip_address = mappings[0][3]

		mappings_without_lease_times = [ mapping[:-1] for mapping in mappings ]

		self.assertIn( ( description_tcp, internal_client, internal_port, external_ip_address, external_port, 'TCP' ), mappings_without_lease_times )
		self.assertIn( ( description_udp, internal_client, internal_port, external_ip_address, external_port, 'UDP' ), mappings_without_lease_times )

		lib.Networking.NATPunch.RemoveUPnPMapping( external_port, 'TCP' )
		lib.Networking.NATPunch.RemoveUPnPMapping( external_port, 'UDP' )

		mappings = lib.Networking.NATPunch.GetUPnPMappings()

		mappings_without_lease_times = [ mapping[:-1] for mapping in mappings ]

		self.assertNotIn( ( description_tcp, internal_client, internal_port, external_ip_address, external_port, 'TCP' ), mappings_without_lease_times )
		self.assertNotIn( ( description_udp, internal_client, internal_port, external_ip_address, external_port, 'UDP' ), mappings_without_lease_times )