File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/CloudFlare.tar
Back
tests/test_purge_cache.py 0000644 00000004673 15030530462 0011574 0 ustar 00 """ get/post/delete/etc zone ruleset based tests """ import os import sys import uuid import random sys.path.insert(0, os.path.abspath('.')) sys.path.insert(0, os.path.abspath('..')) import CloudFlare # test purge_cache cf = None def test_cloudflare(debug=False): """ test_cloudflare """ global cf cf = CloudFlare.CloudFlare(debug=debug) assert isinstance(cf, CloudFlare.CloudFlare) zone_name = None zone_id = None zone_type = None def test_find_zone(domain_name=None): """ test_find_zone """ global zone_name, zone_id, zone_type # grab a random zone identifier from the first 10 zones if domain_name: params = {'per_page':1, 'name':domain_name} else: params = {'per_page':10} try: zones = cf.zones.get(params=params) except CloudFlare.exceptions.CloudFlareAPIError as e: print('%s: Error %d=%s' % (domain_name, int(e), str(e)), file=sys.stderr) assert False assert len(zones) > 0 and len(zones) <= 10 n = random.randrange(len(zones)) zone_name = zones[n]['name'] zone_id = zones[n]['id'] zone_type = zones[0]['plan']['name'] assert len(zone_id) == 32 print('zone: %s %s' % (zone_id, zone_name), file=sys.stderr) def create_purge_zone_data(): """ create_purge_zone_data """ if 'Enterprise' in zone_type: # Enterprise accounts can do all things ... fake_tag = 'tag-' + str(uuid.uuid1()) data = { # 'purge_everything': True, 'hosts': [zone_name], 'tags': [fake_tag], 'prefixes': [zone_name + '/' + 'index.html'], } else: # Free, Pro, Business accounts can only do this ... data = { 'purge_everything': True } return data def test_purge_cache_post(): """ test_purge_cache_post """ r = cf.zones.purge_cache.post(zone_id, data=create_purge_zone_data()) assert isinstance(r, dict) assert 'id' in r assert r['id'] == zone_id def test_purge_cache_delete(): """ test_purge_cache_delete """ # delete method is not in documents; however, it works r = cf.zones.purge_cache.delete(zone_id, data=create_purge_zone_data()) assert isinstance(r, dict) assert 'id' in r assert r['id'] == zone_id if __name__ == '__main__': test_cloudflare(debug=True) if len(sys.argv) > 1: test_find_zone(sys.argv[1]) else: test_find_zone() test_purge_cache_post() test_purge_cache_delete() tests/test_cloudflare_calls.py 0000644 00000013537 15030530462 0012624 0 ustar 00 """ class calling tests """ import os import sys sys.path.insert(0, os.path.abspath('.')) sys.path.insert(0, os.path.abspath('..')) import CloudFlare # test Cloudflare init param (ie. debug, raw, etc) cf = None def test_cloudflare(): """ test_cloudflare """ global cf cf = CloudFlare.CloudFlare() assert isinstance(cf, CloudFlare.CloudFlare) def test_percent_s(): """ test_percent_s """ s = '%s' % cf assert len(s) > 0 and isinstance(s, str) def test_percent_r(): """ test_percent_r """ s = '%r' % cf assert len(s) > 0 and isinstance(s, str) def test_percent_ips_s(): """ test_percent_ips_s """ s = '%s' % cf.ips assert len(s) > 0 and isinstance(s, str) def test_percent_ips_r(): """ test_percent_ips_r """ s = '%r' % cf.ips assert len(s) > 0 and isinstance(s, str) def test_percent_cf_accounts_billing_s(): """ test_percent_cf_accounts_billing_s """ s = '%s' % cf.accounts.billing assert len(s) > 0 and isinstance(s, str) def test_percent_cf_accounts_billing_r(): """ test_percent_cf_accounts_billing_r """ s = '%r' % cf.accounts.billing assert len(s) > 0 and isinstance(s, str) def test_percent_cf_zones_waiting_rooms_events_details_s(): """ test_percent_cf_accounts_billing_s """ s = '%s' % cf.zones.waiting_rooms.events.details assert len(s) > 0 and isinstance(s, str) def test_percent_cf_zones_waiting_rooms_events_details_r(): """ test_percent_cf_accounts_billing_r """ s = '%r' % cf.zones.waiting_rooms.events.details assert len(s) > 0 and isinstance(s, str) def test_ips1(): """ test_ips1 """ ips = cf.ips() assert isinstance(ips, dict) assert len(ips) > 0 def test_cloudflare_debug(): """ test_cloudflare_debug """ global cf cf = CloudFlare.CloudFlare(debug=True) assert isinstance(cf, CloudFlare.CloudFlare) def test_ips2(): """ test_ips2 """ ips = cf.ips() assert isinstance(ips, dict) assert len(ips) > 0 def test_cloudflare_raw(): """ test_cloudflare_raw """ global cf cf = CloudFlare.CloudFlare(raw=False) assert isinstance(cf, CloudFlare.CloudFlare) def test_ips3(): """ test_ips3 """ ips = cf.ips() assert isinstance(ips, dict) assert len(ips) > 0 def test_cloudflare_no_sessions(): """ test_cloudflare_no_sessions """ global cf cf = CloudFlare.CloudFlare(use_sessions=False) assert isinstance(cf, CloudFlare.CloudFlare) def test_ips4(): """ test_ips4 """ ips = cf.ips() assert isinstance(ips, dict) assert len(ips) > 0 def test_ips5(): """ test_ips5 """ ips = cf.ips() assert isinstance(ips, dict) assert len(ips) > 0 def test_cloudflare_url_invalid(): """ test_cloudflare_url_invalid """ global cf cf = CloudFlare.CloudFlare(base_url='blah blah blah blah ...') # this does not fail yet - so we wait def test_ips6_should_fail(): """ test_ips6_should_fail """ try: ips = cf.ips() except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s %d %s' % (type(e).__name__, int(e), str(e)), file=sys.stderr) pass except Exception as e: print('Error expected: %s %s' % (type(e).__name__, e), file=sys.stderr) pass def test_cloudflare_url_wrong(): """ test_cloudflare_url_wrong """ global cf cf = CloudFlare.CloudFlare(base_url='http://example.com/') # this does not fail yet - so we wait def test_ips7_should_fail(): """ test_ips7_should_fail """ try: ips = cf.ips() except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s %d %s' % (type(e).__name__, int(e), str(e)), file=sys.stderr) pass except Exception as e: print('Error expected: %s %s' % (type(e).__name__, e), file=sys.stderr) pass def test_cloudflare_email_invalid(): """ test_cloudflare_email_invalid """ global cf try: cf = CloudFlare.CloudFlare(email=int(0)) assert False except TypeError as e: print('Error expected: %s' % (e), file=sys.stderr) def test_cloudflare_key_invalid(): """ test_cloudflare_key_invalid """ global cf try: cf = CloudFlare.CloudFlare(key=int(0)) assert False except TypeError as e: print('Error expected: %s' % (e), file=sys.stderr) def test_cloudflare_token_invalid(): """ test_cloudflare_token_invalid """ global cf try: cf = CloudFlare.CloudFlare(token=int(0)) assert False except TypeError as e: print('Error expected: %s' % (e), file=sys.stderr) def test_cloudflare_certtoken_invalid(): """ test_cloudflare_certtoken_invalid """ global cf try: cf = CloudFlare.CloudFlare(certtoken=int(0)) assert False except TypeError as e: print('Error expected: %s' % (e), file=sys.stderr) def test_cloudflare_context(): """ test_cloudflare_context """ global cf cf = None with CloudFlare.CloudFlare() as cf: assert isinstance(cf, CloudFlare.CloudFlare) ips = cf.ips() assert isinstance(ips, dict) assert len(ips) > 0 if __name__ == '__main__': test_cloudflare() test_percent_s() test_percent_r() test_percent_ips_s() test_percent_ips_r() test_percent_cf_accounts_billing_s() test_percent_cf_accounts_billing_r() test_percent_cf_zones_waiting_rooms_events_details_s() test_percent_cf_zones_waiting_rooms_events_details_r() test_ips1() test_cloudflare_debug() test_ips2() test_cloudflare_raw() test_ips3() test_cloudflare_no_sessions() test_ips4() test_ips5() test_cloudflare_url_wrong() test_ips6_should_fail() test_cloudflare_url_invalid() test_ips7_should_fail() test_cloudflare_email_invalid() test_cloudflare_key_invalid() test_cloudflare_token_invalid() test_cloudflare_certtoken_invalid() test_cloudflare_context() tests/test_urlscanner.py 0000644 00000007360 15030530462 0011477 0 ustar 00 """ urlscanner tests - PNG data retured """ import os import sys import random import tempfile sys.path.insert(0, os.path.abspath('.')) sys.path.insert(0, os.path.abspath('..')) import CloudFlare cf = None def test_cloudflare(debug=False): """ test_cloudflare """ global cf cf = CloudFlare.CloudFlare(debug=debug) assert isinstance(cf, CloudFlare.CloudFlare) account_name = None account_id = None def test_find_account(find_name=None): """ test_find_account """ global account_name, account_id # grab a random account identifier from the first 10 accounts if find_name: params = {'per_page':1, 'name':find_name} else: params = {'per_page':10} try: accounts = cf.accounts.get(params=params) except CloudFlare.exceptions.CloudFlareAPIError as e: print('%s: Error %d=%s' % (find_name, int(e), str(e)), file=sys.stderr) assert False assert len(accounts) > 0 and len(accounts) <= 10 # n = random.randrange(len(accounts)) # stop using a random account - use the primary account (i.e. the zero'th one) n = 0 account_name = accounts[n]['name'] account_id = accounts[n]['id'] assert len(account_id) == 32 print('account: %s %s' % (account_id, account_name), file=sys.stderr) scan_uuid = None scan_url = None def test_urlscanner(): """ test_urlscanner """ global scan_uuid, scan_url scans = cf.accounts.urlscanner.scan.get(account_id, params={'limit':10}) assert isinstance(scans, dict) assert 'tasks' in scans assert isinstance(scans['tasks'], list) tasks = scans['tasks'] for task in tasks: assert isinstance(task, dict) assert 'success' in task assert 'time' in task assert 'uuid' in task assert 'url' in task print('%s: %s %s %s' % (task['uuid'], task['success'], task['time'], task['url']), file=sys.stderr) n = random.randrange(len(tasks)) scan_uuid = tasks[n]['uuid'] scan_url = tasks[n]['url'] def test_urlscanner_scan(): """ test_urlscanner_scan """ scan = cf.accounts.urlscanner.scan.get(account_id, scan_uuid) assert isinstance(scan, dict) assert 'scan' in scan assert isinstance(scan['scan'], dict) assert 'task' in scan['scan'] task = scan['scan']['task'] assert 'success' in task assert 'time' in task assert 'url' in task assert 'uuid' in task assert 'visibility' in task print('%s: %s %s %s' % (task['uuid'], task['success'], task['time'], task['url']), file=sys.stderr) # https://www.w3.org/TR/png/#5PNG-file-signature # PNG signature 89 50 4E 47 0D 0A 1A 0A def ispng(s): """ ispng """ if b'\x89PNG\x0d\x0a\x1a\x0a' == s[0:8]: return True return False # we don't write out the image - we have no interest in doing this def write_png_file(s): """ write_png_file """ hostname = scan_url.split('/')[2].replace('.', '_') with tempfile.NamedTemporaryFile(mode='wb', prefix='screenshot-' + hostname + '-', suffix='.png', delete=False) as fp: fp.write(s) print('%s' % (fp.name), file=sys.stderr) def test_urlscanner_scan_screenshot(): """ test_urlscanner_scan_screenshot """ # the real test - returning bytes as this is an image png_content = cf.accounts.urlscanner.scan.screenshot.get(account_id, scan_uuid) assert isinstance(png_content, bytes) print('%s: %s png_content: len=%d sig="%s"' % (scan_uuid, scan_url, len(png_content), png_content[0:8]), file=sys.stderr) assert ispng(png_content) # write_png_file(png_content) if __name__ == '__main__': test_cloudflare(debug=True) if len(sys.argv) > 1: test_find_account(sys.argv[1]) else: test_find_account() test_urlscanner() test_urlscanner_scan() test_urlscanner_scan_screenshot() tests/test_graphql.py 0000644 00000015102 15030530462 0010752 0 ustar 00 """ graphql tests """ import os import sys import random import datetime import pytz import json sys.path.insert(0, os.path.abspath('.')) sys.path.insert(0, os.path.abspath('..')) import CloudFlare # test /graphql cf = None def test_cloudflare(debug=False): """ test_cloudflare """ global cf cf = CloudFlare.CloudFlare(debug=debug) assert isinstance(cf, CloudFlare.CloudFlare) zone_name = None zone_id = None def test_find_zone(domain_name=None): """ test_find_zone """ global zone_name, zone_id # grab a random zone identifier from the first 10 zones if domain_name: params = {'per_page':1, 'name':domain_name} else: params = {'per_page':10} try: zones = cf.zones.get(params=params) except CloudFlare.exceptions.CloudFlareAPIError as e: print('%s: Error %d=%s' % (domain_name, int(e), str(e)), file=sys.stderr) assert False assert len(zones) > 0 and len(zones) <= 10 n = random.randrange(len(zones)) zone_name = zones[n]['name'] zone_id = zones[n]['id'] assert len(zone_id) == 32 print('zone: %s %s' % (zone_id, zone_name), file=sys.stderr) def rfc3339_iso8601_time(hour_delta=0, with_hms=False): # format time (with an hour offset in RFC3339 or ISO8601 format (and do it UTC time) if sys.version_info[:3][0] <= 3 and sys.version_info[:3][1] <= 10: dt = datetime.datetime.utcnow().replace(microsecond=0, tzinfo=pytz.UTC) else: dt = datetime.datetime.now(datetime.UTC).replace(microsecond=0) dt += datetime.timedelta(hours=hour_delta) if with_hms: return dt.isoformat().replace('+00:00', 'Z') return dt.strftime('%Y-%m-%d') def test_graphql_get(): """ /graphql_get test """ try: # graphql is alwatys a post - this should fail (but presently doesn't) r = cf.graphql.get() if r is not None and 'data' in r and 'errors' in r: # still an invalid API! print('Error in API (but proceeding) r=', r, file=sys.stderr) assert True else: assert False except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s' % (e), file=sys.stderr) assert True def test_graphql_patch(): """ /graphql_patch test """ try: # graphql is alwatys a post - this should fail (but presently doesn't) r = cf.graphql.patch() if r is not None and 'data' in r and 'errors' in r: # still an invalid API! print('Error in API (but proceeding) r=', r, file=sys.stderr) assert True else: assert False except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s' % (e), file=sys.stderr) assert True def test_graphql_put(): """ /graphql_put test """ try: # graphql is alwatys a post - this should fail (but presently doesn't) r = cf.graphql.put() if r is not None and 'data' in r and 'errors' in r: # still an invalid API! print('Error in API (but proceeding) r=', r, file=sys.stderr) assert True else: assert False except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s' % (e), file=sys.stderr) assert True def test_graphql_delete(): """ /graphql_delete test """ try: # graphql is alwatys a post - this should fail (but presently doesn't) r = cf.graphql.delete() if r is not None and 'data' in r and 'errors' in r: # still an invalid API! print('Error in API (but proceeding) r=', r, file=sys.stderr) assert True else: assert False except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s' % (e), file=sys.stderr) assert True def test_graphql_post_empty(): """ /graphql_post_empty test """ try: # graphql requires data - this should fail (but presently doesn't) r = cf.graphql.post(data={}) if r is not None and 'data' in r and 'errors' in r: # still an invalid API! print('Error in API (but proceeding) r=', r, file=sys.stderr) assert True else: assert False except CloudFlare.exceptions.CloudFlareAPIError as e: print('Error expected: %s' % (e), file=sys.stderr) assert True def test_graphql_post(): """ /graphql_post test """ date_before = rfc3339_iso8601_time(0) # now date_after = rfc3339_iso8601_time(-3 * 24) # 3 days worth query = """ query { viewer { zones(filter: {zoneTag: "%s"} ) { httpRequests1dGroups(limit:40, filter:{date_lt: "%s", date_gt: "%s"}) { sum { countryMap { bytes, requests, clientCountryName } } dimensions { date } } } } } """ % (zone_id, date_before, date_after) # remove whitespace from query - this isn't needed; but helps debug query = '\n'.join([s.strip() for s in query.splitlines()]).strip() # graphql query is always a post try: r = cf.graphql.post(data={'query':query}) except CloudFlare.exceptions.CloudFlareAPIError as e: print('%s: Error %d=%s' % ('/graphql.post', int(e), str(e)), file=sys.stderr) assert False # success - lets confirm it's graphql results as-per query above # basic graphql results assert 'data' in r assert 'errors' in r assert r['errors'] is None # viewer and zones from above assert 'viewer' in r['data'] assert 'zones' in r['data']['viewer'] # only one zone zones = r['data']['viewer']['zones'] assert len(zones) == 1 zone_info = zones[0] # the data assert 'httpRequests1dGroups' in zone_info httpRequests1dGroups = zone_info['httpRequests1dGroups'] assert isinstance(httpRequests1dGroups, list) for h in httpRequests1dGroups: assert 'dimensions' in h assert 'date' in h['dimensions'] result_date = h['dimensions']['date'] assert 'sum' in h result_sum = h['sum'] assert 'countryMap' in result_sum countryMap = h['sum']['countryMap'] for element in countryMap: assert 'bytes' in element assert 'requests' in element assert 'clientCountryName' in element if __name__ == '__main__': test_cloudflare(debug=True) if len(sys.argv) > 1: test_find_zone(sys.argv[1]) else: test_find_zone() test_graphql_get() test_graphql_put() test_graphql_patch() test_graphql_delete() test_graphql_post_empty() test_graphql_post() tests/test_dns_import_export.py 0000644 00000007305 15030530462 0013101 0 ustar 00 """ dns import/export test """ import os import sys import uuid import time import random import tempfile sys.path.insert(0, os.path.abspath('.')) sys.path.insert(0, os.path.abspath('..')) import CloudFlare # test IMPORT EXPORT cf = None def test_cloudflare(): """ test_cloudflare """ global cf cf = CloudFlare.CloudFlare() assert isinstance(cf, CloudFlare.CloudFlare) zone_name = None zone_id = None def test_find_zone(domain_name=None): """ test_find_zone """ global zone_name, zone_id # grab a random zone identifier from the first 10 zones if domain_name: params = {'per_page':1, 'name':domain_name} else: params = {'per_page':10} try: zones = cf.zones.get(params=params) except CloudFlare.exceptions.CloudFlareAPIError as e: print('%s: Error %d=%s' % (domain_name, int(e), str(e)), file=sys.stderr) assert False assert len(zones) > 0 and len(zones) <= 10 n = random.randrange(len(zones)) zone_name = zones[n]['name'] zone_id = zones[n]['id'] assert len(zone_id) == 32 print('zone: %s %s' % (zone_id, zone_name), file=sys.stderr) def test_dns_import(): """ test_dns_import """ # IMPORT # create a zero length file fp = tempfile.TemporaryFile(mode='w+b') fp.seek(0) while True: try: results = cf.zones.dns_records.import_.post(zone_id, files={'file':fp}) break except CloudFlare.exceptions.CloudFlareAPIError as e: print('cf.zones.dns_records.import: returned %d "%s"' % (int(e), str(e))) # 429 or 99998 time.sleep(.5) # This is sadly needed as import seems to be rate limited fp.seek(0) # {"recs_added": 0, "recs_added_by_type": {}, "total_records_parsed": 0} assert len(results) > 0 assert results['recs_added'] == 0 assert len(results['recs_added_by_type']) == 0 assert results['total_records_parsed'] == 0 def test_dns_export(): """ test_dns_export """ # EXPORT dns_records = cf.zones.dns_records.export.get(zone_id) assert len(dns_records) > 0 assert isinstance(dns_records, str) assert 'SOA' in dns_records assert 'NS' in dns_records def test_cloudflare_with_debug(): """ test_cloudflare_with_debug """ global cf cf = CloudFlare.CloudFlare(debug=True) assert isinstance(cf, CloudFlare.CloudFlare) def test_dns_import_with_debug(): """ test_dns_import_with_debug """ # IMPORT # create a zero length file fp = tempfile.TemporaryFile(mode='w+b') fp.seek(0) while True: try: results = cf.zones.dns_records.import_.post(zone_id, files={'file':fp}) break except CloudFlare.exceptions.CloudFlareAPIError as e: print('cf.zones.dns_records.import: returned %d "%s"' % (int(e), str(e))) # 429 or 99998 time.sleep(.5) # This is sadly needed as import seems to be rate limited fp.seek(0) # {"recs_added": 0, "recs_added_by_type": {}, "total_records_parsed": 0} assert len(results) > 0 assert results['recs_added'] == 0 assert len(results['recs_added_by_type']) == 0 assert results['total_records_parsed'] == 0 def test_dns_export_with_debug(): """ test_dns_export_with_debug """ # EXPORT dns_records = cf.zones.dns_records.export.get(zone_id) assert len(dns_records) > 0 assert isinstance(dns_records, str) assert 'SOA' in dns_records assert 'NS' in dns_records if __name__ == '__main__': test_cloudflare() if len(sys.argv) > 1: test_find_zone(sys.argv[1]) else: test_find_zone() test_dns_import() test_dns_export() test_cloudflare_with_debug() test_dns_import_with_debug() test_dns_export_with_debug() tests/__pycache__/test_radar_returning_csv.cpython-310.pyc 0000644 00000003105 15030530462 0017714 0 ustar 00 o �h� � @ s� d Z ddlZddlZddlZej�dej�d�� ej�dej�d�� ddlZdaddd�Z da dd � Zd d� Ze dkrJe d d� e� e� dS dS )z radar returning CSV test � N�.z..Fc C s t j | d�attt j �sJ �dS )z test_cloudflare ��debugN)� CloudFlare�cf� isinstancer � r �\/usr/local/CyberCP/lib/python3.10/site-packages/CloudFlare/tests/test_radar_returning_csv.py�test_cloudflare s r c C sz ddi} t jj| d�}t|�dksJ �d|v sJ �g a|d D ]}t�|d |d |d d f� qttd d� dd �adS )z test_radar_datasets_ranking �limit� )�paramsr �datasets�id�alias�meta�topc S s | d S )N� r )�vr r r �<lambda>$ s z-test_radar_datasets_ranking.<locals>.<lambda>F)�key�reverseN)r �radarr �len�aliases�append�sorted)r �resultsr r r r �test_radar_datasets_ranking s �r c C sn t dd� D ].} | d }| d }tj�|�}t|�dksJ �|�� }|d dks*J �t|�|d ks4J �qdS )z) test_radar_datasets_ranking_two_aliases r r � �domainN)r r r r r � splitlines)r r �n_linesr �linesr r r �'test_radar_datasets_ranking_two_aliases&