-
Notifications
You must be signed in to change notification settings - Fork 141
/
Copy pathffi.py
7377 lines (5382 loc) · 216 KB
/
ffi.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Python bindings for the Pact FFI.
This module provides a Python interface to the Pact FFI. It is a thin wrapper
around the C API, and is intended to be used by the Pact Python client library
to provide a Pythonic interface to Pact.
This module is not intended to be used directly by Pact users. Pact users should
use the Pact Python client library instead. No guarantees are made about the
stability of this module's API.
## Developer Notes
This modules should provide the following only:
- Basic Enum classes
- Simple wrappers around functions, including the casting of input and output
values between the high level Python types and the low level C types.
- Simple wrappers around some of the low-level types. Specifically designed to
automatically handle the freeing of memory when the Python object is
destroyed.
These low-level functions may then be combined into higher level classes and
modules. Ideally, all code outside of this module should be written in pure
Python and not worry about allocating or freeing memory.
During initial implementation, a lot of these functions will simply raise a
`NotImplementedError`.
For those unfamiliar with CFFI, please make sure to read the [CFFI
documentation](https://cffi.readthedocs.io/en/latest/using.html).
### Handles
The Rust library exposes a number of handles to internal data structures. This
is done to avoid exposing the internal implementation details of the library to
users of the library, and avoid unnecessarily casting to and from possibly
complicated structs.
In the Rust library, the handles are thin wrappers around integers, and
unfortunately the CFFI interface sees this and automatically unwraps them,
exposing the underlying integer. As a result, we must re-wrap the integer
returned by the CFFI interface. This unfortunately means that we may be subject
to changes in private implementation details upstream.
### Freeing Memory
Python has a garbage collector, and as a result, we don't need to worry about
manually freeing memory. Having said that, Python's garbace collector is only
aware of Python objects, and not of any memory allocated by the Rust library.
To ensure that the memory allocated by the Rust library is freed, we must make
sure to define the
[`__del__`](https://docs.python.org/3/reference/datamodel.html#object.__del__)
method to call the appropriate free function whenever the Python object is
destroyed.
Note that there are some rather subtle details as to when this is called, when
it may never be called, and what global variables are accessible. This is
explained in the documentation for `__del__` above, and in Python's [garbage
collection](https://docs.python.org/3/library/gc.html) module.
### Error Handling
The FFI function should handle all errors raised by the function call, and raise
an appropriate Python exception. The exception should be raised using the
appropriate Python exception class, and should be documented in the function's
docstring.
"""
# The following lints are disabled during initial development and should be
# removed later.
# ruff: noqa: ARG001 (unused-function-argument)
# ruff: noqa: A002 (builtin-argument-shadowing)
# ruff: noqa: D101 (undocumented-public-class)
# The following lints are disabled for this file.
# ruff: noqa: SLF001
# private-member-access, as we need access to other handles' internal
# references, without exposing them to the user.
from __future__ import annotations
import gc
import json
import logging
import typing
import warnings
from enum import Enum
from typing import TYPE_CHECKING, Any, List
from pact.v3._ffi import ffi, lib # type: ignore[import]
if TYPE_CHECKING:
import datetime
from collections.abc import Collection
from pathlib import Path
import cffi
from typing_extensions import Self
logger = logging.getLogger(__name__)
# The follow types are classes defined in the Rust code. Ultimately, a Python
# alternative should be implemented, but for now, the follow lines only serve
# to inform the type checker of the existence of these types.
class AsynchronousMessage: ...
class Consumer: ...
class Generator: ...
class GeneratorCategoryIterator: ...
class GeneratorKeyValuePair: ...
class HttpRequest: ...
class HttpResponse: ...
class InteractionHandle:
"""
Handle to a HTTP Interaction.
[Rust
`InteractionHandle`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/mock_server/handles/struct.InteractionHandle.html)
"""
def __init__(self, ref: int) -> None:
"""
Initialise a new Interaction Handle.
Args:
ref:
Reference to the Interaction Handle.
"""
self._ref: int = ref
def __str__(self) -> str:
"""
String representation of the Interaction Handle.
"""
return f"InteractionHandle({self._ref})"
def __repr__(self) -> str:
"""
String representation of the Interaction Handle.
"""
return f"InteractionHandle({self._ref!r})"
class MatchingRule: ...
class MatchingRuleCategoryIterator: ...
class MatchingRuleDefinitionResult: ...
class MatchingRuleIterator: ...
class MatchingRuleKeyValuePair: ...
class MatchingRuleResult: ...
class Message: ...
class MessageContents: ...
class MessageHandle: ...
class MessageMetadataIterator: ...
class MessageMetadataPair: ...
class MessagePact: ...
class MessagePactHandle: ...
class MessagePactMessageIterator: ...
class MessagePactMetadataIterator: ...
class MessagePactMetadataTriple: ...
class Mismatch: ...
class Mismatches: ...
class MismatchesIterator: ...
class Pact: ...
class PactHandle:
"""
Handle to a Pact.
[Rust
`PactHandle`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/mock_server/handles/struct.PactHandle.html)
"""
def __init__(self, ref: int) -> None:
"""
Initialise a new Pact Handle.
Args:
ref:
Rust library reference to the Pact Handle.
"""
self._ref: int = ref
def __del__(self) -> None:
"""
Destructor for the Pact Handle.
"""
cleanup_plugins(self)
free_pact_handle(self)
def __str__(self) -> str:
"""
String representation of the Pact Handle.
"""
return f"PactHandle({self._ref})"
def __repr__(self) -> str:
"""
String representation of the Pact Handle.
"""
return f"PactHandle({self._ref!r})"
class PactServerHandle:
"""
Handle to a Pact Server.
This does not have an exact correspondance in the Rust library. It is used
to manage the lifecycle of the mock server.
# Implementation Notes
The Rust library uses the port number as a unique identifier, in much the
same was as it uses a wrapped integer for the Pact handle.
"""
def __init__(self, ref: int) -> None:
"""
Initialise a new Pact Server Handle.
Args:
ref:
Rust library reference to the Pact Server.
"""
self._ref: int = ref
def __del__(self) -> None:
"""
Destructor for the Pact Server Handle.
"""
cleanup_mock_server(self)
def __str__(self) -> str:
"""
String representation of the Pact Server Handle.
"""
return f"PactServerHandle({self._ref})"
def __repr__(self) -> str:
"""
String representation of the Pact Server Handle.
"""
return f"PactServerHandle({self._ref!r})"
@property
def port(self) -> int:
"""
Port on which the Pact Server is running.
"""
return self._ref
class PactInteraction: ...
class PactInteractionIterator:
"""
Iterator over a Pact's interactions.
Interactions encompasses all types of interactions, including HTTP
interactions and messages.
"""
def __init__(self, ptr: cffi.FFI.CData) -> None:
"""
Initialise a new Pact Interaction Iterator.
Args:
ptr:
CFFI data structure.
"""
if ffi.typeof(ptr).cname != "struct PactInteractionIterator *":
msg = (
"ptr must be a struct PactInteractionIterator, got"
f" {ffi.typeof(ptr).cname}"
)
raise TypeError(msg)
self._ptr = ptr
def __str__(self) -> str:
"""
Nice string representation.
"""
return "PactInteractionIterator"
def __repr__(self) -> str:
"""
Debugging representation.
"""
return f"PactInteractionIterator({self._ptr!r})"
def __del__(self) -> None:
"""
Destructor for the Pact Interaction Iterator.
"""
pact_interaction_iter_delete(self)
def __next__(self) -> PactInteraction:
"""
Get the next interaction from the iterator.
"""
return pact_interaction_iter_next(self)
class PactMessageIterator:
"""
Iterator over a Pact's asynchronous messages.
"""
def __init__(self, ptr: cffi.FFI.CData) -> None:
"""
Initialise a new Pact Message Iterator.
Args:
ptr:
CFFI data structure.
"""
if ffi.typeof(ptr).cname != "struct PactMessageIterator *":
msg = (
f"ptr must be a struct PactMessageIterator, got {ffi.typeof(ptr).cname}"
)
raise TypeError(msg)
self._ptr = ptr
def __str__(self) -> str:
"""
Nice string representation.
"""
return "PactMessageIterator"
def __repr__(self) -> str:
"""
Debugging representation.
"""
return f"PactMessageIterator({self._ptr!r})"
def __del__(self) -> None:
"""
Destructor for the Pact Message Iterator.
"""
pact_message_iter_delete(self)
def __iter__(self) -> Self:
"""
Return the iterator itself.
"""
return self
def __next__(self) -> Message:
"""
Get the next message from the iterator.
"""
return pact_message_iter_next(self)
class PactSyncHttpIterator:
"""
Iterator over a Pact's synchronous HTTP interactions.
"""
def __init__(self, ptr: cffi.FFI.CData) -> None:
"""
Initialise a new Pact Synchronous HTTP Iterator.
Args:
ptr:
CFFI data structure.
"""
if ffi.typeof(ptr).cname != "struct PactSyncHttpIterator *":
msg = (
"ptr must be a struct PactSyncHttpIterator, got"
f" {ffi.typeof(ptr).cname}"
)
raise TypeError(msg)
self._ptr = ptr
def __str__(self) -> str:
"""
Nice string representation.
"""
return "PactSyncHttpIterator"
def __repr__(self) -> str:
"""
Debugging representation.
"""
return f"PactSyncHttpIterator({self._ptr!r})"
def __del__(self) -> None:
"""
Destructor for the Pact Synchronous HTTP Iterator.
"""
pact_sync_http_iter_delete(self)
def __iter__(self) -> Self:
"""
Return the iterator itself.
"""
return self
def __next__(self) -> SynchronousHttp:
"""
Get the next message from the iterator.
"""
return pact_sync_http_iter_next(self)
class PactSyncMessageIterator:
"""
Iterator over a Pact's synchronous messages.
"""
def __init__(self, ptr: cffi.FFI.CData) -> None:
"""
Initialise a new Pact Synchronous Message Iterator.
Args:
ptr:
CFFI data structure.
"""
if ffi.typeof(ptr).cname != "struct PactSyncMessageIterator *":
msg = (
"ptr must be a struct PactSyncMessageIterator, got"
f" {ffi.typeof(ptr).cname}"
)
raise TypeError(msg)
self._ptr = ptr
def __str__(self) -> str:
"""
Nice string representation.
"""
return "PactSyncMessageIterator"
def __repr__(self) -> str:
"""
Debugging representation.
"""
return f"PactSyncMessageIterator({self._ptr!r})"
def __del__(self) -> None:
"""
Destructor for the Pact Synchronous Message Iterator.
"""
pact_sync_message_iter_delete(self)
def __iter__(self) -> Self:
"""
Return the iterator itself.
"""
return self
def __next__(self) -> SynchronousMessage:
"""
Get the next message from the iterator.
"""
return pact_sync_message_iter_next(self)
class Provider: ...
class ProviderState: ...
class ProviderStateIterator: ...
class ProviderStateParamIterator: ...
class ProviderStateParamPair: ...
class SynchronousHttp: ...
class SynchronousMessage: ...
class VerifierHandle:
"""
Handle to a Verifier.
[Rust `VerifierHandle`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/verifier/handle/struct.VerifierHandle.html)
"""
def __init__(self, ref: cffi.FFI.CData) -> None:
"""
Initialise a new Verifier Handle.
Args:
ref:
Rust library reference to the Verifier Handle.
"""
self._ref = ref
def __del__(self) -> None:
"""
Destructor for the Verifier Handle.
"""
verifier_shutdown(self)
def __str__(self) -> str:
"""
String representation of the Verifier Handle.
"""
return f"VerifierHandle({hex(id(self._ref))})"
def __repr__(self) -> str:
"""
String representation of the Verifier Handle.
"""
return f"<VerifierHandle: _ref at {hex(id(self._ref))}>"
class ExpressionValueType(Enum):
"""
Expression Value Type.
[Rust `ExpressionValueType`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/models/expressions/enum.ExpressionValueType.html)
"""
UNKNOWN = lib.ExpressionValueType_Unknown
STRING = lib.ExpressionValueType_String
NUMBER = lib.ExpressionValueType_Number
INTEGER = lib.ExpressionValueType_Integer
DECIMAL = lib.ExpressionValueType_Decimal
BOOLEAN = lib.ExpressionValueType_Boolean
def __str__(self) -> str:
"""
Informal string representation of the Expression Value Type.
"""
return self.name
def __repr__(self) -> str:
"""
Information-rich string representation of the Expression Value Type.
"""
return f"ExpressionValueType.{self.name}"
class GeneratorCategory(Enum):
"""
Generator Category.
[Rust `GeneratorCategory`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/models/generators/enum.GeneratorCategory.html)
"""
METHOD = lib.GeneratorCategory_METHOD
PATH = lib.GeneratorCategory_PATH
HEADER = lib.GeneratorCategory_HEADER
QUERY = lib.GeneratorCategory_QUERY
BODY = lib.GeneratorCategory_BODY
STATUS = lib.GeneratorCategory_STATUS
METADATA = lib.GeneratorCategory_METADATA
def __str__(self) -> str:
"""
Informal string representation of the Generator Category.
"""
return self.name
def __repr__(self) -> str:
"""
Information-rich string representation of the Generator Category.
"""
return f"GeneratorCategory.{self.name}"
class InteractionPart(Enum):
"""
Interaction Part.
[Rust `InteractionPart`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/mock_server/handles/enum.InteractionPart.html)
"""
REQUEST = lib.InteractionPart_Request
RESPONSE = lib.InteractionPart_Response
def __str__(self) -> str:
"""
Informal string representation of the Interaction Part.
"""
return self.name
def __repr__(self) -> str:
"""
Information-rich string representation of the Interaction Part.
"""
return f"InteractionPath.{self.name}"
class LevelFilter(Enum):
"""Level Filter."""
OFF = lib.LevelFilter_Off
ERROR = lib.LevelFilter_Error
WARN = lib.LevelFilter_Warn
INFO = lib.LevelFilter_Info
DEBUG = lib.LevelFilter_Debug
TRACE = lib.LevelFilter_Trace
def __str__(self) -> str:
"""
Informal string representation of the Level Filter.
"""
return self.name
def __repr__(self) -> str:
"""
Information-rich string representation of the Level Filter.
"""
return f"LevelFilter.{self.name}"
class MatchingRuleCategory(Enum):
"""
Matching Rule Category.
[Rust `MatchingRuleCategory`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/models/matching_rules/enum.MatchingRuleCategory.html)
"""
METHOD = lib.MatchingRuleCategory_METHOD
PATH = lib.MatchingRuleCategory_PATH
HEADER = lib.MatchingRuleCategory_HEADER
QUERY = lib.MatchingRuleCategory_QUERY
BODY = lib.MatchingRuleCategory_BODY
STATUS = lib.MatchingRuleCategory_STATUS
CONTENST = lib.MatchingRuleCategory_CONTENTS
METADATA = lib.MatchingRuleCategory_METADATA
def __str__(self) -> str:
"""
Informal string representation of the Matching Rule Category.
"""
return self.name
def __repr__(self) -> str:
"""
Information-rich string representation of the Matching Rule Category.
"""
return f"MatchingRuleCategory.{self.name}"
class PactSpecification(Enum):
"""
Pact Specification.
[Rust `PactSpecification`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/models/pact_specification/enum.PactSpecification.html)
"""
UNKNOWN = lib.PactSpecification_Unknown
V1 = lib.PactSpecification_V1
V1_1 = lib.PactSpecification_V1_1
V2 = lib.PactSpecification_V2
V3 = lib.PactSpecification_V3
V4 = lib.PactSpecification_V4
@classmethod
def from_str(cls, version: str) -> PactSpecification:
"""
Instantiate a Pact Specification from a string.
This method is case-insensitive, and allows for the version to be
specified with or without a leading "V", and with either a dot or an
underscore as the separator.
Args:
version:
The version of the Pact Specification.
Returns:
The Pact Specification.
"""
version = version.upper().replace(".", "_")
if version.startswith("V"):
return cls[version]
return cls["V" + version]
def __str__(self) -> str:
"""
Informal string representation of the Pact Specification.
"""
return self.name
def __repr__(self) -> str:
"""
Information-rich string representation of the Pact Specification.
"""
return f"PactSpecification.{self.name}"
class StringResult:
"""
String result.
"""
class _StringResult(Enum):
"""
Internal enum from Pact FFI.
[Rust `StringResult`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/mock_server/enum.StringResult.html)
"""
FAILED = lib.StringResult_Failed
OK = lib.StringResult_Ok
class _StringResultCData:
tag: int
ok: cffi.FFI.CData
failed: cffi.FFI.CData
def __init__(self, cdata: cffi.FFI.CData) -> None:
"""
Initialise a new String Result.
Args:
cdata:
CFFI data structure.
"""
if ffi.typeof(cdata).cname != "struct StringResult":
msg = f"cdata must be a struct StringResult, got {ffi.typeof(cdata).cname}"
raise TypeError(msg)
self._cdata = typing.cast(StringResult._StringResultCData, cdata)
def __str__(self) -> str:
"""
String representation of the String Result.
"""
return self.text
def __repr__(self) -> str:
"""
Debugging string representation of the String Result.
"""
return f"<StringResult: {'OK' if self.is_ok else 'FAILED'}, {self.text!r}>"
@property
def is_failed(self) -> bool:
"""
Whether the result is an error.
"""
return self._cdata.tag == StringResult._StringResult.FAILED.value
@property
def is_ok(self) -> bool:
"""
Whether the result is ok.
"""
return self._cdata.tag == StringResult._StringResult.OK.value
@property
def text(self) -> str:
"""
The text of the result.
"""
# The specific `.ok` or `.failed` does not matter.
s = ffi.string(self._cdata.ok)
if isinstance(s, bytes):
return s.decode("utf-8")
return s
def raise_exception(self) -> None:
"""
Raise an exception with the text of the result.
Raises:
RuntimeError: If the result is an error.
"""
if self.is_failed:
raise RuntimeError(self.text)
class OwnedString(str):
"""
A string that owns its own memory.
This is used to ensure that the memory is freed when the string is
destroyed.
As this is subclassed from `str`, it can be used in place of a normal string
in most cases.
"""
__slots__ = ("_ptr", "_string")
def __new__(cls, ptr: cffi.FFI.CData) -> Self:
"""
Create a new Owned String.
As this is a subclass of the immutable type `str`, we need to override
the `__new__` method to ensure that the string is initialised correctly.
"""
s = ffi.string(ptr)
return super().__new__(cls, s if isinstance(s, str) else s.decode("utf-8"))
def __init__(self, ptr: cffi.FFI.CData) -> None:
"""
Initialise a new Owned String.
Args:
ptr:
CFFI data structure.
"""
self._ptr = ptr
s = ffi.string(ptr)
self._string = s if isinstance(s, str) else s.decode("utf-8")
def __str__(self) -> str:
"""
String representation of the Owned String.
"""
return self._string
def __repr__(self) -> str:
"""
Debugging string representation of the Owned String.
"""
return f"<OwnedString: {self._string!r}, ptr={self._ptr!r}>"
def __del__(self) -> None:
"""
Destructor for the Owned String.
"""
string_delete(self)
def __eq__(self, other: object) -> bool:
"""
Equality comparison.
Args:
other:
The object to compare to.
Returns:
Whether the two objects are equal.
"""
if isinstance(other, OwnedString):
return self._ptr == other._ptr
if isinstance(other, str):
return self._string == other
return super().__eq__(other)
def version() -> str:
"""
Return the version of the pact_ffi library.
[Rust `pactffi_version`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/?search=pactffi_version)
Returns:
The version of the pact_ffi library as a string, in the form of `x.y.z`.
"""
v = ffi.string(lib.pactffi_version())
if isinstance(v, bytes):
return v.decode("utf-8")
return v
def init(log_env_var: str) -> None:
"""
Initialise the mock server library.
This can provide an environment variable name to use to set the log levels.
This function should only be called once, as it tries to install a global
tracing subscriber.
[Rust
`pactffi_init`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/?search=pactffi_init)
# Safety
log_env_var must be a valid NULL terminated UTF-8 string.
"""
raise NotImplementedError
def init_with_log_level(level: str = "INFO") -> None:
"""
Initialises logging, and sets the log level explicitly.
This function should only be called once, as it tries to install a global
tracing subscriber.
[Rust
`pactffi_init_with_log_level`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/?search=pactffi_init_with_log_level)
Args:
level:
One of TRACE, DEBUG, INFO, WARN, ERROR, NONE/OFF. Case-insensitive.
# Safety
Exported functions are inherently unsafe.
"""
raise NotImplementedError
def enable_ansi_support() -> None:
"""
Enable ANSI coloured output on Windows.
On non-Windows platforms, this function is a no-op.
[Rust
`pactffi_enable_ansi_support`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/?search=pactffi_enable_ansi_support)
# Safety
This function is safe.
"""
raise NotImplementedError
def log_message(
message: str,
log_level: LevelFilter | str = LevelFilter.ERROR,
source: str | None = None,
) -> None:
"""
Log using the shared core logging facility.
[Rust
`pactffi_log_message`](https://docs.rs/pact_ffi/0.4.18/pact_ffi/?search=pactffi_log_message)
This is useful for callers to have a single set of logs.
Args:
message:
The contents written to the log
log_level:
The verbosity at which this message should be logged.
source:
The source of the log, such as the class, module or caller.
"""
if isinstance(log_level, str):
log_level = LevelFilter[log_level.upper()]
if source is None:
import inspect
source = inspect.stack()[1].function