Satisfy pre-commit checks

program_admin/__init__.py

@@ -3,7 +3,7 @@
 import sys
 from dataclasses import asdict
 from pathlib import Path
-from typing import Any, Dict, List, Literal, Tuple, Optional
+from typing import Any, Dict, List, Literal, Optional, Tuple
 
 from loguru import logger
 from solana import system_program
@@ -160,9 +160,11 @@ async def refresh_program_accounts(self):
  logger.debug(f"Found {len(self._price_accounts)} price account(s)")
 
  if self.authority_permission_account:
- logger.debug(f"Found permission account: {self.authority_permission_account.data}")
+ logger.debug(
+ f"Found permission account: {self.authority_permission_account.data}"
+ )
  else:
- logger.debug(f"Authority permission account not found")
+ logger.debug("Authority permission account not found")
 
  async def send_transaction(
  self, instructions: List[TransactionInstruction], signers: List[Keypair]
@@ -289,16 +291,25 @@ async def sync(
 
  if ref_authority_permissions:
  # Sync authority permissions
- (authority_instructions, authority_signers) = await self.sync_authority_permissions_instructions(ref_authority_permissions)
+ (
+ authority_instructions,
+ authority_signers,
+ ) = await self.sync_authority_permissions_instructions(
+ ref_authority_permissions
+ )
 
  if authority_instructions:
  instructions.extend(authority_instructions)
 
  if send_transactions:
- await self.send_transaction(authority_instructions, authority_signers)
+ await self.send_transaction(
+ authority_instructions, authority_signers
+ )
 
  else:
- logger.debug("Reference data for authority permissions is not defined, skipping...")
+ logger.debug(
+ "Reference data for authority permissions is not defined, skipping..."
+ )
 
  return instructions
 
@@ -541,15 +552,22 @@ async def sync_authority_permissions_instructions(
  ) -> Tuple[List[TransactionInstruction], List[Keypair]]:
  instructions = []
  signers = []
- if (not self.authority_permission_account or
- not self.authority_permission_account.matches_reference_data(reference_authority_permissions)):
- upgrade_authority_keypair = load_keypair("upgrade_authority", key_dir=self.key_dir)
+ if (
+ not self.authority_permission_account
+ or not self.authority_permission_account.matches_reference_data(
+ reference_authority_permissions
+ )
+ ):
+ upgrade_authority_keypair = load_keypair(
+ "upgrade_authority", key_dir=self.key_dir
+ )
 
  logger.debug("Building pyth_program.upd_permissions instruction")
- instruction = pyth_program.upd_permissions(self.program_key,
- upgrade_authority_keypair.public_key,
- reference_authority_permissions,
- )
+ instruction = pyth_program.upd_permissions(
+ self.program_key,
+ upgrade_authority_keypair.public_key,
+ reference_authority_permissions,
+ )
  instructions = [instruction]
  signers = [upgrade_authority_keypair]
  else:

program_admin/cli.py

@@ -389,7 +389,9 @@ def restore_links(network, rpc_endpoint, program_key, keys, products, commitment
  "--permissions", help="Path to reference permissions file", envvar="PERMISSIONS"
 )
 @click.option(
- "--authority-permissions", help="Path to reference authority permissions file", envvar="AUTHORITY_PERMISSIONS"
+ "--authority-permissions",
+ help="Path to reference authority permissions file",
+ envvar="AUTHORITY_PERMISSIONS",
 )
 @click.option(
  "--overrides", help="Path to reference overrides file", envvar="OVERRIDES"
@@ -442,8 +444,10 @@ def sync(
  ref_authority_permissions = None
 
  if authority_permissions:
- ref_authority_permissions = parse_authority_permissions_json(Path(authority_permissions))
-
+ ref_authority_permissions = parse_authority_permissions_json(
+ Path(authority_permissions)
+ )
+
  asyncio.run(
  program_admin.sync(
  ref_products=ref_products,

program_admin/instructions.py

@@ -2,12 +2,11 @@
 
 from construct import Bytes, Int32sl, Int32ul, Struct
 from solana.publickey import PublicKey
-from solana.transaction import AccountMeta, TransactionInstruction
 from solana.system_program import SYS_PROGRAM_ID
-
-from program_admin.util import encode_product_metadata
+from solana.transaction import AccountMeta, TransactionInstruction
 
 from program_admin.types import ReferenceAuthorityPermissions
+from program_admin.util import encode_product_metadata
 
 # TODO: Implement add_mapping instruction
 
@@ -25,11 +24,12 @@
 PRICE_TYPE_PRICE = 1
 PROGRAM_VERSION = 2
 
-AUTHORITY_PERMISSIONS_PDA_SEED = b"permissions";
+AUTHORITY_PERMISSIONS_PDA_SEED = b"permissions"
 
 # NOTE(2023-07-11): currently the loader's address is not part of our version of solana-py
 BPF_UPGRADEABLE_LOADER_ID = "BPFLoaderUpgradeab1e11111111111111111111111"
 
+
 def init_mapping(
  program_key: PublicKey, funding_key: PublicKey, mapping_key: PublicKey
 ) -> TransactionInstruction:
@@ -265,11 +265,12 @@ def toggle_publisher(
  program_id=program_key,
  )
 
+
 def upd_permissions(
-  program_key: PublicKey,
-  upgrade_authority: PublicKey,
-  refdata: ReferenceAuthorityPermissions,
- ) -> TransactionInstruction:
+ program_key: PublicKey,
+ upgrade_authority: PublicKey,
+ refdata: ReferenceAuthorityPermissions,
+) -> TransactionInstruction:
  """
  Pyth program upd_permissions instruction. Sets contents of the
  permission account which allows us to name various authorities:
@@ -292,16 +293,17 @@ def upd_permissions(
  "master_authority" / Bytes(32),
  "data_curation_authority" / Bytes(32),
  "security_authority" / Bytes(32),
-  )
+ )
 
  ix_data = ix_data_layout.build(
- dict(version=PROGRAM_VERSION,
-  command=COMMAND_UPD_PERMISSIONS,
-  master_authority=bytes(refdata["master_authority"]),
-  data_curation_authority=bytes(refdata["data_curation_authority"]),
-  security_authority=bytes(refdata["security_authority"]),
- )
+ dict(
+ version=PROGRAM_VERSION,
+ command=COMMAND_UPD_PERMISSIONS,
+ master_authority=bytes(refdata["master_authority"]),
+ data_curation_authority=bytes(refdata["data_curation_authority"]),
+  security_authority=bytes(refdata["security_authority"]),
  )
+ )
 
  [permissions_account, _bump] = PublicKey.find_program_address(
  [AUTHORITY_PERMISSIONS_PDA_SEED],
@@ -319,10 +321,12 @@ def upd_permissions(
  return TransactionInstruction(
  data=ix_data,
  keys=[
- AccountMeta(pubkey=upgrade_authority, is_signer=True, is_writable=True),
- AccountMeta(pubkey=oracle_program_data_key, is_signer=False, is_writable=False),
- AccountMeta(pubkey=permissions_account, is_signer=False, is_writable=True),
- AccountMeta(pubkey=SYS_PROGRAM_ID, is_signer=False, is_writable=False),
- ],
+ AccountMeta(pubkey=upgrade_authority, is_signer=True, is_writable=True),
+ AccountMeta(
+ pubkey=oracle_program_data_key, is_signer=False, is_writable=False
+ ),
+ AccountMeta(pubkey=permissions_account, is_signer=False, is_writable=True),
+ AccountMeta(pubkey=SYS_PROGRAM_ID, is_signer=False, is_writable=False),
+ ],
  program_id=program_key,
-  )
+ )

program_admin/parsing.py

@@ -21,16 +21,14 @@
  PythMappingAccount,
  PythPriceAccount,
  PythProductAccount,
- ReferenceOverrides,
  ReferenceAuthorityPermissions,
+ ReferenceOverrides,
  ReferencePermissions,
  ReferenceProduct,
  ReferencePublishers,
 )
 from program_admin.util import apply_overrides
 
-from loguru import logger
-
 MAGIC_NUMBER = "0xa1b2c3d4"
 VERSION = 2
 
@@ -159,22 +157,25 @@ def parse_price_data(data: bytes) -> PriceData:
 
 
 def parse_authority_permission_data(data: bytes) -> AuthorityPermissionData:
- # Start by offsetting the header, currently 4 * u32 = 16 bytes 
- data_with_current_offset = data[16:] 
- 
+ # Start by offsetting the header, currently 4 * u32 = 16 bytes
+ data_with_current_offset = data[16:]
+
  master_authority = PublicKey(data_with_current_offset[:32])
 
  # Continue adjusting offset after parsing each chunk of the data.
  data_with_current_offset = data_with_current_offset[32:]
 
  data_curation_authority = PublicKey(data_with_current_offset[:32])
  data_with_current_offset = data_with_current_offset[32:]
- 
+
  security_authority = PublicKey(data_with_current_offset[:32])
 
- return AuthorityPermissionData(master_authority, data_curation_authority, security_authority)
+ return AuthorityPermissionData(
+ master_authority, data_curation_authority, security_authority
+ )
 
 
+# pylint: disable=too-many-return-statements
 def parse_data(data: bytes) -> Optional[AccountData]:
  magic_number = hex(Int32ul.parse(data[0:]))
  version = Int32ul.parse(data[4:])
@@ -196,7 +197,7 @@ def parse_data(data: bytes) -> Optional[AccountData]:
  return parse_authority_permission_data(data)
  if data_type == ACCOUNT_TYPE_TEST:
  return None
- 
+
  raise RuntimeError(f"Invalid account type: {data_type}")
 
 
@@ -240,19 +241,23 @@ def parse_publishers_json(file_path: Path) -> ReferencePublishers:
  "names": names,
  }
 
+
 def parse_permissions_json(file_path: Path) -> ReferencePermissions:
  with file_path.open() as stream:
  return json.load(stream)
 
+
 def parse_authority_permissions_json(file_path: Path) -> ReferenceAuthorityPermissions:
  with file_path.open() as stream:
  # Vanilla Python does not enforce type hints, explicitly build pubkeys
  perm_dict = json.load(stream)
 
  return ReferenceAuthorityPermissions(
  master_authority=PublicKey(str(perm_dict["master_authority"])),
- data_curation_authority=PublicKey(str(perm_dict["data_curation_authority"])),
- security_authority=PublicKey(str(perm_dict["security_authority"]))
+ data_curation_authority=PublicKey(
+ str(perm_dict["data_curation_authority"])
+ ),
+ security_authority=PublicKey(str(perm_dict["security_authority"])),
  )
 
 

program_admin/types.py

@@ -35,6 +35,7 @@ class ReferenceAuthorityPermissions(TypedDict):
  data_curation_authority: PublicKey
  security_authority: PublicKey
 
+
 # network -> symbol -> enabled / disabled. Default is no change to permissions.
 ReferenceOverrides = Dict[str, Dict[str, bool]]
 
@@ -114,6 +115,7 @@ class PriceData:
  def __str__(self) -> str:
  return f"PriceData(product_key={str(self.product_account_key)[0:5]}...)"
 
+
 @dataclass
 class AuthorityPermissionData:
  master_authority: PublicKey
@@ -154,6 +156,7 @@ class PythProductAccount(PythAccount):
 class PythPriceAccount(PythAccount):
  data: PriceData
 
+
 @dataclass
 class PythAuthorityPermissionAccount(PythAccount):
  """
@@ -164,9 +167,12 @@ class PythAuthorityPermissionAccount(PythAccount):
  account is responsible for global oracle administration
  authorities.
  """
+
  data: AuthorityPermissionData
 
  def matches_reference_data(self, refdata: ReferenceAuthorityPermissions) -> bool:
- return (refdata["master_authority"] == self.data.master_authority and
- refdata["data_curation_authority"] == self.data.data_curation_authority and
- refdata["security_authority"] == self.data.security_authority)
+ return (
+ refdata["master_authority"] == self.data.master_authority
+ and refdata["data_curation_authority"] == self.data.data_curation_authority
+ and refdata["security_authority"] == self.data.security_authority
+ )