23753134726

Committed 30 Mar 2026 03:33PM UTC coverage: 91.985% (+3.4%) from 88.611%

Build # 23753134726

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #222 from Nanopublication/fix/nanopub-creation

Mainly bug fixing for the Nanopub creation from an existing nanopub

Coverage Stats

303 of 312 new or added lines in 17 files covered. (97.12%)

42 existing lines in 3 files now uncovered.

3397 of 3693 relevant lines covered (91.98%)

1.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.62

/tests/java_wrapper.py

import os
import subprocess
import tempfile
from base64 import decodebytes
from pathlib import Path

from Crypto.PublicKey import RSA
from rdflib import Dataset, RDF

from nanopub.definitions import ROOT_FILEPATH
from nanopub.namespaces import NP
from nanopub.nanopub import Nanopub
from nanopub.profile import PROFILE_INSTRUCTIONS_MESSAGE

# nanopub-java is only used in dev or tests when the repo is cloned
NANOPUB_JAVA_SCRIPT = (ROOT_FILEPATH / 'scripts' / 'nanopub-java')


class JavaWrapper:
    """
    Wrapper around 'nanopub-java' java tool that is used to sign and publish nanopublications to
    a nanopub server.
    """

    def __init__(self, private_key: str = None) -> None:
        """Construct JavaWrapper.

        Args:
            use_test_server: Toggle using the test nanopub server.
        """
        if private_key:
            # Put keys in files (needed for nanopub-java)
            keys_dir = tempfile.mkdtemp()
            private_key_path = os.path.join(keys_dir, "id_rsa")
            with open(private_key_path, "w") as f:
                f.write(private_key + '\n')
            self.private_key = str(private_key_path)

            public_key_path = os.path.join(keys_dir, "id_rsa.pub")
            key = RSA.import_key(decodebytes(private_key.encode()))
            public_key = key.publickey().export_key().decode('utf-8').replace("-----BEGIN PUBLIC KEY-----\n",
                                                                              "").replace("-----END PUBLIC KEY-----",
                                                                                          "")
            with open(public_key_path, "w") as f:
                f.write(public_key)

    def _run_command(self, command):
        result = subprocess.run(command, shell=True, stderr=subprocess.PIPE)
        rsa_key_messages = ['FileNotFoundException', 'id_rsa']
        stderr = result.stderr.decode('utf8')
        if all(m in stderr for m in rsa_key_messages):
            raise RuntimeError('Nanopub RSA key appears to be missing,\n'
                               + PROFILE_INSTRUCTIONS_MESSAGE
                               + '\nDetailed error message:\n' + stderr)
        elif all(m in stderr for m in ['SignatureException', 'Seems to have signature']):
            raise RuntimeError('The Publication you are trying to publish already has a signature, '
                               'this means it is likely already published. '
                               'If you want to publish a modified existing nanopublication '
                               'you need to do a few extra steps before you can publish. '
                               'See the discussion in: '
                               'https://github.com/Nanopublication/nanopub-py/issues/110')
        elif result.returncode != 0:
            raise RuntimeError(f'Error in nanopub-java when running {command}: {stderr}')

    def sign(self, np: Nanopub) -> str:
        tmp_dir = tempfile.mkdtemp()
        unsigned_file = os.path.join(tmp_dir, "unsigned.trig")
        with open(unsigned_file, "w") as f:
            f.write(np.rdf.serialize(format="trig"))

        args = f'-k {self.private_key}' if self.private_key else ''
        cmd = f'{NANOPUB_JAVA_SCRIPT} sign {unsigned_file} {args}'
        self._run_command(cmd)

        signed_file = self._get_signed_file(unsigned_file)
        if not os.path.exists(signed_file):
            raise RuntimeError(f"Signed file not created: {signed_file}")

        g = Dataset()
        g.parse(signed_file, format="trig")

        np_uris = [s for s, p, o, c in g.quads((None, RDF.type, NP.Nanopublication, None))]
        if not np_uris:
            raise ValueError("No nanopublication found in signed file.")
        return str(np_uris[0])

    def check_trusty_with_signature(self, np: Nanopub) -> str:
        tmp_dir = tempfile.mkdtemp()
        np_file = os.path.join(tmp_dir, "signed.trig")
        with open(np_file, "w") as f:
            f.write(np.rdf.serialize(format="trig"))
        np_file = str(np_file)

        cmd = f'{NANOPUB_JAVA_SCRIPT} check {np_file}'
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        print(str(result.stdout))
        return "1 trusty with signature" in str(result.stdout)

    def _get_signed_file(self, unsigned_file: str):
        unsigned_path = Path(unsigned_file)
        return str(unsigned_path.parent / f'signed.{unsigned_path.name}')

1	import os	2✔
2	import subprocess	2✔
3	import tempfile	2✔
4	from base64 import decodebytes	2✔
5	from pathlib import Path	2✔
6
7	from Crypto.PublicKey import RSA	2✔
8	from rdflib import Dataset, RDF	2✔
9
10	from nanopub.definitions import ROOT_FILEPATH	2✔
11	from nanopub.namespaces import NP	2✔
12	from nanopub.nanopub import Nanopub	2✔
13	from nanopub.profile import PROFILE_INSTRUCTIONS_MESSAGE	2✔
14
15	# nanopub-java is only used in dev or tests when the repo is cloned
16	NANOPUB_JAVA_SCRIPT = (ROOT_FILEPATH / 'scripts' / 'nanopub-java')	2✔
17
18
19	class JavaWrapper:	2✔
20	"""
21	Wrapper around 'nanopub-java' java tool that is used to sign and publish nanopublications to
22	a nanopub server.
23	"""
24
25	def __init__(self, private_key: str = None) -> None:	2✔
26	"""Construct JavaWrapper.
27
28	Args:
29	use_test_server: Toggle using the test nanopub server.
30	"""
31	if private_key:	2✔
32	# Put keys in files (needed for nanopub-java)
33	keys_dir = tempfile.mkdtemp()	2✔
34	private_key_path = os.path.join(keys_dir, "id_rsa")	2✔
35	with open(private_key_path, "w") as f:	2✔
36	f.write(private_key + '\n')	2✔
37	self.private_key = str(private_key_path)	2✔
38
39	public_key_path = os.path.join(keys_dir, "id_rsa.pub")	2✔
40	key = RSA.import_key(decodebytes(private_key.encode()))	2✔
41	public_key = key.publickey().export_key().decode('utf-8').replace("-----BEGIN PUBLIC KEY-----\n",	2✔
42	"").replace("-----END PUBLIC KEY-----",
43	"")
44	with open(public_key_path, "w") as f:	2✔
45	f.write(public_key)	2✔
46
47	def _run_command(self, command):	2✔
UNCOV 48	result = subprocess.run(command, shell=True, stderr=subprocess.PIPE)	×
49	rsa_key_messages = ['FileNotFoundException', 'id_rsa']	×
50	stderr = result.stderr.decode('utf8')	×
51	if all(m in stderr for m in rsa_key_messages):	×
52	raise RuntimeError('Nanopub RSA key appears to be missing,\n'	×
53	+ PROFILE_INSTRUCTIONS_MESSAGE
54	+ '\nDetailed error message:\n' + stderr)
55	elif all(m in stderr for m in ['SignatureException', 'Seems to have signature']):	×
56	raise RuntimeError('The Publication you are trying to publish already has a signature, '	×
57	'this means it is likely already published. '
58	'If you want to publish a modified existing nanopublication '
59	'you need to do a few extra steps before you can publish. '
60	'See the discussion in: '
61	'https://github.com/Nanopublication/nanopub-py/issues/110')
62	elif result.returncode != 0:	×
63	raise RuntimeError(f'Error in nanopub-java when running {command}: {stderr}')	×
64
65	def sign(self, np: Nanopub) -> str:	2✔
UNCOV 66	tmp_dir = tempfile.mkdtemp()	×
67	unsigned_file = os.path.join(tmp_dir, "unsigned.trig")	×
68	with open(unsigned_file, "w") as f:	×
69	f.write(np.rdf.serialize(format="trig"))	×
70
71	args = f'-k {self.private_key}' if self.private_key else ''	×
72	cmd = f'{NANOPUB_JAVA_SCRIPT} sign {unsigned_file} {args}'	×
73	self._run_command(cmd)	×
74
75	signed_file = self._get_signed_file(unsigned_file)	×
76	if not os.path.exists(signed_file):	×
77	raise RuntimeError(f"Signed file not created: {signed_file}")	×
78
79	g = Dataset()	×
80	g.parse(signed_file, format="trig")	×
81
NEW 82	np_uris = [s for s, p, o, c in g.quads((None, RDF.type, NP.Nanopublication, None))]	×
83	if not np_uris:	×
84	raise ValueError("No nanopublication found in signed file.")	×
UNCOV 85	return str(np_uris[0])	×
86
87	def check_trusty_with_signature(self, np: Nanopub) -> str:	2✔
UNCOV 88	tmp_dir = tempfile.mkdtemp()	×
89	np_file = os.path.join(tmp_dir, "signed.trig")	×
90	with open(np_file, "w") as f:	×
91	f.write(np.rdf.serialize(format="trig"))	×
92	np_file = str(np_file)	×
93
94	cmd = f'{NANOPUB_JAVA_SCRIPT} check {np_file}'	×
95	result = subprocess.run(cmd, shell=True, capture_output=True, text=True)	×
96	print(str(result.stdout))	×
97	return "1 trusty with signature" in str(result.stdout)	×
98
99	def _get_signed_file(self, unsigned_file: str):	2✔
UNCOV 100	unsigned_path = Path(unsigned_file)	×
101	return str(unsigned_path.parent / f'signed.{unsigned_path.name}')	×

Nanopublication / nanopub-py / 23753134726

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous