Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_multiple_devices_data_message(push_service):
try:
push_service.multiple_devices_data_message(
data_message={"test": "Test"},
dry_run=True
)
assert False, "Should raise InvalidDataError without registration ids"
except errors.InvalidDataError:
pass
response = push_service.multiple_devices_data_message(
registration_ids=["Test"],
data_message={"test": "Test"},
dry_run=True
)
assert isinstance(response, dict)
def test_init_baseapi():
try:
BaseAPI()
assert False, "Should raise AuthenticationError"
except errors.AuthenticationError:
pass
def test_push_service_without_credentials():
try:
FCMNotification()
assert False, "Should raise AuthenticationError without credentials"
except errors.AuthenticationError:
pass
def base_api():
api_key = os.getenv("FCM_TEST_API_KEY", None)
assert api_key, "Please set the environment variables for testing according to CONTRIBUTING.rst"
return BaseAPI(api_key=api_key)
def test_init_baseapi():
try:
BaseAPI()
assert False, "Should raise AuthenticationError"
except errors.AuthenticationError:
pass
def test_notify_topic_subscribers(push_service):
try:
push_service.notify_topic_subscribers(
message_body="Test",
dry_run=True
)
assert False, "Should raise InvalidDataError without topic"
except errors.InvalidDataError:
pass
response = push_service.notify_topic_subscribers(
topic_name="test",
message_body="Test",
message_title="Test",
dry_run=True
)
assert isinstance(response, dict)
def __init__(self, conf, router_conf, metrics):
"""Create a new FCM router and connect to FCM"""
self.conf = conf
self.router_conf = router_conf
self.metrics = metrics
self.min_ttl = router_conf.get("ttl", 60)
self.dryRun = router_conf.get("dryrun", False)
self.collapseKey = router_conf.get("collapseKey", "webpush")
self.senderID = router_conf.get("senderID")
self.auth = router_conf.get("auth")
self._base_tags = ["platform:fcm"]
try:
self.fcm = pyfcm.FCMNotification(api_key=self.auth)
except Exception as e:
self.log.error("Could not instantiate FCM {ex}",
ex=e)
raise IOError("FCM Bridge not initiated in main")
self.log.debug("Starting FCM router...")
"databaseURL": databaseURL,
"storageBucket": ""
}
firebase = pyrebase.initialize_app(config_pyrebase)
mAuth = firebase.auth()
user = mAuth.sign_in_with_email_and_password(mail, password)
userID = mAuth.get_account_info(user['idToken'])
userID = userID["users"][0]["localId"]
db = firebase.database()
db_user = db.child("users").child(userID).get(user['idToken'])
deviceID = db_user.val()
push_service = FCMNotification(api_key=apiKeyFCM)
conn = sqlite3.connect('/opt/kripton-guard/network.db')
def createTables(conn):
#Create db table if it's not exist
conn.execute("CREATE TABLE mac_ip_addresses (ID INTEGER PRIMARY KEY AUTOINCREMENT, macAddress varchar(17) UNIQUE NOT NULL, ipAddress varchar(15) NOT NULL, comment varchar(50) )")
def showDevices():
#Shows devices in whitelist
print "\n========== Your Whitelist ==========\n Mac Address IP Address "
query = "SELECT macAddress,ipAddress FROM mac_ip_addresses;"
result = conn.execute(query)
for row in result:
print row[0] + " " + row[1] + "\n"
print "==================================="
from pyfcm import FCMNotification
# fcm 푸쉬 메시지는 data message를 전송할 수 있다
fcm = FCMNotification(api_key='')
data = {'key1': 123, 'key2': 1234}
fcm.notify_single_device(data_message=data)
click_action=None,
badge=None,
color=None,
tag=None,
body_loc_key=None,
body_loc_args=None,
title_loc_key=None,
title_loc_args=None,
content_available=None,
timeout=5,
extra_notification_kwargs=None,
extra_kwargs={}):
if api_key is None:
api_key = SETTINGS.get("FCM_SERVER_KEY")
push_service = FCMNotification(api_key=api_key, json_encoder=json_encoder)
result = push_service.notify_topic_subscribers(
topic_name=topic_name,
message_body=message_body,
message_title=message_title,
message_icon=message_icon,
sound=sound,
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
restricted_package_name=restricted_package_name,
low_priority=low_priority,
dry_run=dry_run,
data_message=data_message,
click_action=click_action,
badge=badge,