From 8d0b849143c44ae431e8808bc5ad6ba9255b083f Mon Sep 17 00:00:00 2001 From: Devin Slick Date: Thu, 6 Nov 2025 10:06:13 -0600 Subject: [PATCH] Bump version to 2.0.2 and add extensive unit tests to improve code coverage to 98%. The tests cover various edge cases and error handling in client.py and device.py, including authentication flows, connection error retries, JSON parsing, and response handling. Additional tests ensure robustness against invalid inputs and unexpected server responses. --- .coverage | Bin 53248 -> 81920 bytes coverage.xml | 624 +++++------ fmd_api/_version.py | 2 +- tests/unit/test_coverage_improvements.py | 1221 ++++++++++++++++++++++ 4 files changed, 1540 insertions(+), 307 deletions(-) create mode 100644 tests/unit/test_coverage_improvements.py diff --git a/.coverage b/.coverage index 0fa58e2c2e801b0861ec08aa850ec6298f14186d..e251d6dd82ef7f809dee83e86508a0233a7cd0f6 100644 GIT binary patch literal 81920 zcmeI52b>gD_V4e#Rn>LtbRy?4faEZeGm?XnqaZL0!;ohNW`-QQySl{yCd7<_h^_&% zq9S6>30(}UuDZ&itFCKab)9$4^u3q%|Nrm%_}=@!@ZVSSA^g5wRXtUwyU*!!&#iN( zOd7kSvN&&HS$RoeWnOnygK^FV=jAcRJp5Y+|LPwXg{=Yq=Q;YHNso0ow@x%Zokc?1 zS$uAELcDY2>S&wrtC14#-0*I<&?~|X!B1d;zyg5<0t@{AZGpt3h+Ds5L!R7JSvY@r zabJkvDb7(6J-(vXAApgFO7{+&OPZUU}J?ycNaec?*{;FV0(1 zx?o9BVdaw2MR}Eri}l{CDvB3Spu@*7)X|Pzs*fjCuw(%qRav|UpJGM%l9Iymb$QE* z*LAAiLH)MM;FD@-AuE?H6yCnBu z&Cf2bzNW(RBK#Bn)z#nwCog5pLVTj4vhoFV_VddNON$m4S9HoNG*?(uhUdAqGJCb< zWhMEQWqHe&lvZD6X~mMtC1s^~#cPX;R#g@+_zy3z6B$kapz(6A2A^vlUC_Mjk?MEl zSm)%y|G|-JcXjYWJh{>`ZPt?q(yzvYtIxS~Rmprb`O;M-E9k|+BeK7^vh1H7`Ol86 zEK?Wzn_rv%;I%w?y}x=nwaMuf%^tO?bjiw9#pcEDly~yTPSuxDTGlC#E&|^Ht;3=E zeQR@u@5QE?ZwHR~4s zZ@-c7t<%*D)gRIPpS^X=TcDt;k&tK@a_cv4%#&XA+ep1y>Oc2aZz26k{s(SUZz?G+ ztXNg99m#*sN$Mj`dP)5mQCskGI=i9zLmL0H%TfCkbX7toT-&W*w=Pd)KM~}Y=&$pu zi=anNegytaAo%U4zoM{mao)c_cX}94ysZ=T@1ZF}6Sef946@t#;#QLzNWDuu0ze)`e8wu=iZ$O~1p&6{6Vw!FBoG#eZ6 zh%puTJC{`CRiDGZk4mSVe2$8B^c;&;x2--W9bCxGPsIm%v^LJT+Hlb%NDO`~KxYLW0^Ned27yrAb z%SNN&g_XsXOG=9KXqW1fseWUvDXhpVDlf*37)#r?wf=i=1A4Uzy3)UtB}@7$muSx2 z`W-v+#47b+%=(#v3JmHch5ED=ieJ4+{>{dd@0SVaaPiwt?)!vnmtzjX>~A3FmVE{1 zXRV<3Qr~r;eH71<$A4n@C-@005Lh6vKwyEu0)Yhr3j`JjED%^Aus~pezyg5<{`D+i zalu{6|3&d5SRk-KV1d8_fdv8!1Qz(0wt$861)@~{ z=kcEz{t12p3j`JjED%^Aus~pezyg5<0t*Bd2rLj-Ah1ASfqy{@SXR`d`ai4jFL-gm zB?T4;ED%^Aus~pezyg5<0t*Bd2rLj-Ah1ASfxrUB0#yGG@_%EFKm-;DED%^Aus~pe zzyg5<0t*Bd2rLj-Ah1ASfqzL0ICxbe{taDQ?f8%JZ{n}TpNk)k-xc2Bf@@(Xh$i0zUB6}j2MYcy$k#&)>NO9z>$fU@q z$iPVVNc%|hNZm*{!o%N(KM(&Y{8sqo@RQ+(!gq#m3||wzBz#VIOL$dyS-3DfBRnoV zEZjGoA8s9P9IhF5;UDldd;))j*Wp<>40pp#a1C4p8Q1_TpcrPs1Q-r|p$oKv1`q|| zeeZqdec&DQp7)M;_j)&b*Lpj>bG*%7g}2z7<4yKPdi}i4UJI|TC*@D_3;BV3T|O-z zl()(2b~o~;y&))=l;&U+P%O{x~ts9?pf}5 zcZl1=ZR<92qb>`59XcL*EA&F>keyRH4!71sII7OUJUw$8N9um)LO zty8S}7Knd{FU6n4G4ZT8ByJab#ie4K*dR(pA%kQmhkHT@-h{7%WJd$dAlbn{BS^N_ zP$eMQPC;Yd1d?qPG~$gR*+xM_j>oiCP@gx1|RJrQE)dq1c}K8?t#Q4 z1$WUeov7eWb{8ZjD7b??0*Ubo4pc3G#5e`Fu>+78tKe348zjz9a0|N?5~nM;ncV`3 zF$!*CH$!5yf*aXQkQk+4Kf4hUBNg1h_CsQXf_>};NDNnSJ=+I~VG8!L>me~z1D)Ft z1$)?DNDNkRHM;{6rzyCsYB?kZDcHp>gTz1um$F@u7@&c6>#txZyBZSx6kJ@j0TO)` zT*NMhL>~ngu!|tkTfq*3UIs3JL{A0V*$znbP;fp$cLUoY(M`d5?0iUcRd6oBsRqu2 zM81M^*twACqF@_AX9MRzLd73;HroaX6@OTUoec>Ue^`dDN5vnOW;?w^b^PIirFMIX z>hOa)xf>EH`mh8`dI^d?T!<|!;Uy^epx(?iY+AZQHm@mOG;&e;tEN8 zl%fjCPN)ZL+vgI;C@q{EkO3?)6(q%G1v4kW(N|6NRVw4m|P%c7AQ3T~e zloUfy78iL5iXdppB9@j33LyA!5uPQ550ZFl6g)`c`B3N}i6=pUgCzcD3L7MsN{Shz z7E1~kq!vkv7NizR3KgV^CB+F+^CbldRPz^0iV&oVB!vf5^Y%!J4X6s{O9~9A=IoOc z6;RDCkQ5S7&3IK(JU}&VhNNJCYU(sekpR_{sgl9~s>xF%#Q;8$&BSuPc^r(i9kmTi24I3`W z#iJTJOp<>`HRM%E&K=d@p|Vh^A(GrWJ~l94l21o9V4x(2PAXrLH%HZffFxH=DqoTx zN7b*tBqvTPUy=t$)wiD{_f0BalJ7>`W(O)6iK*GAR5k0h5(DqoVnM%AmgBxg-3 zUy`Rr)w7o*H%%&El8;8!qo*VXO)6iKcShAMUy^G^)iqy|Uq+QbTar^IHBgdACN)5k zJ0{g%k}oFJPm&`h)mM@iCe=rh3ntZDlK&;uOOo>?)l&{ps)r=EOOJJv>|n0qH5n!l9xr*uDv7|i>hrqN&Xd8o3@giE2`FQBzaa; zEn7=+tEgJEl;l%UHE$uwp`vQmT#`3M)wG!;SBk1hQ%QalRpTa-oG7YBjU{>6*lDs9V8ns;Vm3YN9+-g7R zi7Hv`B)K@+MJgBh>R_on%}%8e@VJh>;Q6@ zq)WvmKpvBHvA7z@Ws)uw*8%xV($mCkKu(kNSaC0q*Cah$90qcmq( z4M$nz+sLPpKSo~1jQ?=t?#NA%Ya$m%&W>!1tc)zge1BqOM5JFNKXOW>VI&r@!arcP z|6%xz@C)Hb!}o=6311h!G<+`R`jz1&;kn@{;nCrN;cnr!;U<{rhu|mp5R}{B7PwZ-uwO zJJTEM4fc9??Yt&l4bPT8$j{{anB_ks56L^^^>UZoCO61ZIS;e^;j))(CmYF#`-}Uf z`=R@~`;>doy%}@-9qv}Q++E;KcSpPZ+)i#Yw}vZ1-(ZISX6U)lq0oWQb)kzxerR=Q zacEX(Y-kYX_bo$pLLuh|=Tqk$=OyQebCUWBaSSyx_*F#CHBeZqb)go!Zw z-VkQO>^no43bStwVJ^(RF@(u5``Qp@!|bb*gnRzY5az?|OGB6tv%ea`jF^342vcJA zxgpGn*=L3@DQ17kp;y?ahA=H=pBTctn4K_$i81@LAZOqAIhhA>lR#|&Yr%zkeOb7l6rAxxIpYdLfeJ8B5iW%jBe%$M0KhA?4f zFB`&)nZ0BPQ)c#}Ae{HnY82Z`{Q8XuV-0yH4wME7`SL zuU*G>YrSSIyGHBPYuMFVuUgHn(zBLIr~;?IlEw$mNQZ2nxt3$bEz@%Jn59~d8oflzk)sxCIb!4@ zEr*X-sO7NX#aa#>wm{1vuNG-Jc<6j1hZJf#Fn^ww0|pitnLk&{{sZP1nLk_0e*Mof zGJlqqef!NcGXG31`}Cb*Wd3w5d-s`UWd2kwd-a}TWd39=d-j@SWd1}gd-R-OWd3+9 zyXBA5vTOcWE%RreVdTKmjT|t>$o`{^>^I8Dz9Wt7Gs4K;!;S1U%*dWYjqEYR$Zms; z?0Q=D+Zx>$rtpLG!(F-!)Usoj0a~{2*k8+b?fYrjwq0K>+qCVYW$QM*wQSkCmzFJB z_SCX@iym4wYu;VUrp>x(*`#S#EgLsERm(<=^R;ZysEd~M8+6vPUj0s5)~(l3%Q|&C zXj!{Xdo63#Zl`69T5Uc2OjjrQZAg);u4=6}AMJ9Amah(OrRBakEw$V;uZ5PoSLRg* z0kkP)_{|Mqir>rt=J-twV3Oa&0A~4(4Pctz$N=W~4GmzT-@pK7`t=Q9s$b6l=K6It zP_kdg0A~BO4Pd%oO9SQmH4R|GU&8=q{BZ-A^2ZEd&L1^^Nq@uuX8mCUl=d6Iyx%i` ziN7>JnZFmJmxJ>9TFUj>;i%Y+Y_cH!KL3dS0UtS;31zbZd?=XzPeurS0t*Bd2rLj- zAh1ASfxrTR1p*5M76>d5SRk;#zpe%7{{J7XuUP#1_~-FI#ovm*9Mu1V`hQUW#~Rfy zJgEQE>DDe|I^zcsQ=TX|Npf5zgrbjGXWgeQOb5~r7YW0N?1Zx&lvFVR%;K~ z9|f@ZSMiVH$Kub#AB^9MbN?^IJO3-=3*%?TPmd44`~LIdwc>W{JDmCdcI<`N;n*Fq zy?D=mCbkym{m+h#mp|jZ?(fT^@(FpLyh&au&zGC!3OQd+l_T*^_YSg&j7jEx<$mNI zbDwb^bZ>QcyBFep?v?IB_e}S6cYxc)&2wwHcIdm%iO}14m;2$+9ihFUouN!IzKv}Iqx~IIFC8^IQyN;opYUyPAT5uKG_-W^m5ucjhu-6i~Xhj zq5Zo3lzkBIZ@=2!VQ;m|?FII9d$irp?qoN!YuEzsZa;3lX+38hvJP0+Sr=Qrwc1*2 z&9cVgz3r!3Ev-6MNcoW;RpB> z-hr3k2;2qxU>9tI^>{zRT$l($p$D|Vt^?rx?0w^9KH>wxzm#A1W7+rlZadNCRO5j!os7kwrASo9ve-{SJdG;}4uieF*q3Vsd0Jcq92ml?X8 zU%_`7x{P1WFFi^4oI4F&%6IZh4DIBX@r!flQhw1%!s9M9bO|1JfuSAzS-!*2*?6zc zc0(DY^9`kWhM#B1=V$YCb12QvF_gmNwi!zD6hGTgf;ZwBLtA-*rwwi4Te)v&GvC5f zhBoodJZWen-^3G!Ht;0hnnN4;7DMZ}&o}4L2ENJATI{B<(a)%9rsafR`GY&FAnEL$i1VU!Fs=`7%Q@`C7g-hi365hNknWe6gWvd^%rbXeyt^7v|6u zUTkOzpUD^G&{SSzXcC{o=jTupUTA1M+H;4Ff^8r<8uw2!N>ABhQ{#Ge72#{ zd<;J;heq*PhDPx-_{g+VXb1uc0=)E$?HfHE+Xv8#;xz=DiHH;-~PQhFbDg zyhjc-;oS|jz;o$lD37<`U2~`jKh;okp2zbIHRD})mmF%&I~&4YExc0>HRBy~s1fgQ zlF(zeH-!CKcsoPbu!XlZgdJOW8$;N#g|{|@JzMxGhOlW1Z)FI(w(ypQux$%(VF>%S z@H|7=2gq>S>Geg+Ag*QD(xUGpHY~I2fM|LpVJn}yOfWK|%9h^P+mZ7)s?$0+3 zy~*F=Zy0)mzsZjoI>uk&zc=(cKgM4-^csJizm`L<@S}!~^4IvQhF-;IdBxBx{2l&s z4!yx&%AuqD#T$4^a6j8KV#^5{sMp6(6jsmf6CC) z7@eOq^dx_pKVj$zq{nmUN&c9j$N2sHQA0=g6a0vwNB9x`NDkf44;wnnAK?!hdYB*P zhYTI!4|8*G4S$F~#ve4>9^!}i1BM>Ntp^P~z#ruI8#>4z;P)B2AJ6RG96HGFF?26p z!`+7N;rH^p4BgG|;ddIkgWt~YFmyW}cY6-q!4DWZfP3C%=r(>gzcq&r@LLSs!f)d@ z8@id_!hdJzcl>64lc5{&HM-Hze!hqAH?)sm&u=hvJ>SpwYiJL@k?+Z& zz5F^u*Wqut*3h;5I=MQz*8ism{z9Md@CsNE% z57AphbEv0?80v=ELD*2f=qA8WXOS;FL!CrtAq{m91BIJIokYk`d(lBShT4f?!uTw_ zp7z2r+u9-vLv3(RoMWfe-Zx}@Ba__{{{X3g8qL&|34n|{|oy62^RGK z6L{-?(El&!{}=TC3;O>B{r`gge^|E;`v0L3g8qNm-h2P``u}lZxxBg~V3mks{$E3! z%Hm(gkK^qB7vhg#@BbU(m&ebGZ;r2wFNn{;82|(0r^Z`h=YKEubL_9N4`aWNJrjE< zb|AI~=K!1?TOTWl&5KRJp8tJgov`y?t(X)2A^I1rLcfaL{_l_89K8m+{-vU;vD5$T z=!EFdXiuy|H;%@lJn{|p_3@gMu$TW)coGi6EwCFq z`TN-4Zz;^ddipTv1?`~;#IcM2x7gk958g}OquxE`Uy7z0O{47ub{S5!mIgBhGNBY1`KK)~D9H)+^TI)_vCRtgEdHu+QHr zoa1noHQpLx^|0DnjjX7}#Mk1ucndrIJ%Y0wZV;D?^TcMcQY;WNaC~(jzCqLv_|&qL zhGD0RgHOFnvrq{>wJ*&=8Tiz}l%PbEf=^9M36_cy@Trd} z!D6upd}?J%ut+QgpSqc5VKMmB(3D`|KlXx8JxvLU=@@EjnuUenQ)g3x1)>;yYHmtU zBo=^A{Y?qxiz4u;#VJ9dm=8X6IVG4U3c;sFrv!7wdhn^&DZ$KN8h}sjP6^HwGr^~h zrvx*^nc!2?Q-W!<8#O*9m@1}$Pt8vWCX0IDQ~y(fNn$ei)B-gNxP`i)5{wsA{0S=}^YJ&nCONZ2Y1vr!r zsdWmPiabcIHP8%FYZNpQO(C^fL1WPxQmYI!fmEe}M)cGb3L4UGrX}8%5KHwih>MRA=JqJ>=4155onF_Er45>2}U}+dqGZbKD7*f*} zU||?i(-dG`7*bOeU|ASaQxsrT7*dlB;BJ!?U`-fO6BS@d7*Z1yU_}^G;|;t4sc{C5 zLTapmS0QzV0;~o@>U0HI42IMg1y~D))My1*3Wn4u1y~7&)JO$b2!_-M1y~1$)Nln@ z28Pry1y}`!)KCLYLTZQttN}x6umUUrL+UgISOJFAAOpDDKm}O;g_MdvSnfRxDHVUP zzWXqwRQ#dhFQiobK{x~{6@ReQ`w*m5{J~1^gOF13hhyOvQY!vng%(do#UF%&kW%pn ztGswRD*j-R_g+Y;_=C0DdmyFa566-(q*VODnl2vGMO^}xf+3~i565~hq*VODg6`dr zQt=1j0Hjp>!Q$?1kW%pnYrD5VO2r>6r``-H6@NHZdLgCa50-NGKuX0Qj+I{vhmyl!`xC(Z#1$@rPr97g8$zU&M$!7Qt^l5j6z7M_=D5*&W4nVKL}|^srbWjIw7P~{K1)iDM+dKgOdl*_d4oL zv6UwvrQ#1x9oz~j6@PH%;1)=!_=6J%H$zIrACB_~A;r`a;50%=s`$e<@QsjE@du~z z_>ffb2VnywRs6w8bhxyVia$7G5JANsoG`cwk}Ce-JiJOss`!J`1#veOe{ir4e{gEwT1cw+gD?w{D*oW~z^RZ_@dsxI zPKTt5KRAtW8YGLAd8hIzkW}#pCj-ueq>4WX_yj8c;8Z!>O~;=}bUpLb6W}yKNb2~5 zvjicj;}1>}grtr?I7bkYI{x4kK}hQOgEIsnspHRRJ_?dL{^0yTNb2}AijL9o2WJOD zQpX>h90*Are}?lBkks)9r~5%t$Dd(zjE+AzF%XhE{tV&6AgSXIP78#jjz5F>5J>9y zgOdUwspAjM352AMKR6{2k~;q2j6g{0_=6JyA*tgJ&Ig2~jz2gZ5Ry9n^yYmaspAh$ z2884YWnP>M2uU4(a4H}qb^O7ZfRNPjryHG{jz2gL5R!w{6W}yJNb2~L&$~iW#~+*o z2uU4(y6}8R>iC0G03oU456%FDq>ev00T7Zp{^0ySNb2~5)Bhl;;}6dMgQSi>IQb8f zI{x6?KS=8MgH!(?spC&ex&$46aN-{%b^OWWEg-4mPaf^2;}6dIgQSi>IOz|PI{q}{ z&A}^WkBQa)i}U~gSLdI5^Kk9F;a)H7s^7?qVCMa${1AKUKP3;!o8{GVhukX5Xu4S-0=5##QrXxntcy?x}7|w~iYM{Sf*TbMBWy zM{v!&eW6{UZK3s{<)OKuiJ_sP9++`A2m!8`_l5JmbJTglxzD-Dxzah`+2X9ke0!QR z3fIf)=rnaOQrKVHAKP!(&)N@RwtcOA5w4cE%3fs8w9l{y+WB@1ySD9MuKkJi2V5)f z5$jItdTW=p&024jTJx4niifK2jQ)kM|gyjw2)8B5fj#a1{g*{x1A! z_`UGa@KfOj@jipS;a#`}!sc*!cp=_tFg`pq+$-E6+$@ZJA>l{(0zQP_<4T7Q!<~4y z!DVm`u5(xd1uzL$IqW6+Ks#s*QLoDT%KHe{IDAH}eQ>FFuIPmK9F(ixk|AMPdj6+_ z9iKB+6Sc(|*j*ggdcZ*OvDWx`H*I)ceX&Y(?-F?}Hu+ZI?ZTDxPyD#JF6+(Ak zwrvo)`?762;&N&virq(TTU=J%Q`8f7mC(JHaTx@mdoSDHfzZ8|?JsKx-Fw;ow4S(F zpUp>2gzmj;g})QJ_p$|5Lib*_+#iMRy=;X*=-$f~-Y0UtIvdLp=c%)?s)X*mZ2jm8 z-FeyiE-bd`*-_#JT)pFLL(k(FOs2ZKCpEx+u1ciMrl)bq58u#};%SjG^aO588hRWl zVdydOgxH!xkBcpa9u-fD%{la#*ktI4cvNgO^awuZ21AGOc9``!^oUqz=#Y3$tTpri zE?Tn2&_SfthVBy&h*ddsP*fVaSKKEm4BdkZOO_kD6Bk8UY3L556^3pXcZ#wcx;V!ELVaV?u^hAt2nim8TnhzrCN zL+6W&#biV0;(9of44os+6%!3@6X%EthBD%OG2W1mYwCh^y#m2!~vSjxp6^uA;r!hJ&u6ogo}`6>SaSu&Zce2*+JTYeP8jDo!zk zBd?-W4%HJa4dK|UXkiEkUqxOHjTg-g;qa?yW(db$MN>mK04tj0&?wOu*ILR`(Iymf ztxj57A*;FpH^wrIuGy~Jo(NT0?erc$hOD+)e^$rR*XXcIt4b}rM#e&YFkU!pC$hdu zhh4%hwe(dI?X+4dm*cRDk6Zd89d=<=rKK;@!KEQCwzA#0IX>mN+bw;K4#Tw}EPag* zEu&~Eg6@3fv$B)&Czs`y6h$y3DFT2GoRzB2kD@i(m}OcGyeJ#Lct ztJY&Dh%byDCqCDD#3J#T*5=cKB?_V6O2OR#uCWW2E&l7>4Ir?F!T)Zqp{Z!be=vkq z4F0_#EMxHRa;ORa)({pl_&0{IlEJ^up+@{GLs-k;e=~%|4E|*fHQ;|WgyjtWg(0kG z@XrllL4$v02rC-=FF90`e`*M88vGMOSk&MrawyLKYzWI5{J0^kYw(W^VPS)RWC$x8 z{7;6kw81~rM70h6fgvnz@b?X2b%X!W5SBOidxo&S!QVB61rGiW)&E`F`p>NY*N%mv zf5*Gf|A_1UJsCY1y(PLkdQsGmu8A&<&WTRMHUD}=+ee#3<57Vt{+)>Y0rTBQBlkpZ zj9eMnj(5^m;A(#}BjX~2Bi$qIB26Q;BVPEI@YmrJ;de35eH!Ng+!@{X_yruU-vn0G&}_qWH}<(-Ff0#yBd+O}$#4i@gs1 zCXdTM$XDc(@KFC`yS2@c*cF$ zz0183SO2@n&A1!f6>c$(R!qP-0)4UjK`Xa`8^tvMzYl#D`XF>H^gNDP+#9+%bS?Hk zI486@RDml2&JIlqjSTe<<%e2f9|R9aF8=C#=p1vNbq+bVJJ;iCfam1Q6&U1nciKD6 zojQ((-4MREPuTC`nt)H+hwMA;{q|M%g?8FrZ)TO$I|M)f z|5(7y&}gYgD~D!isFZtdmCVpMDd(YvpUDi3lA>(TP-bY16eX?l(;l(hQ>!x z)~zEmG&)KWpMl0kQP!y^Gc+>w zL*t()LsDj_`bN@~8LGXZv~UNiyrIN-QW>hdc_`fs72UWO!YiYS8*cs)_iU4upVfg( zYn1d29KVr6CR=T*{yGksR%#1=8;49w13SG;3j(}b?&9N+$x|Eft2ktuE5L8!UZ$Da z>)h>-X{t8h_i)HGQGkPRkZG&{zmt2JM(U6Z9@0>4z;EP`X`leVkVB@v0{lJ>nR*KF z>o{cUD!^~!UZxHLe!s^r<6fq=)>~z!7HN)O#bu^uR^saBnHpKSexsMcfm1pVzlY0A zEc@`v3uPvnl`AS_hN^4$6s4EQ4As_9maLE&s;r@0wo7KHu7+}HiOf({4dvo1WQJ;L zC>Jf2nNU_1KO{3$Ps2xXIqwWr(@+-9kQpm0=M~BfMp1gSV4lqItejgQGb}6T%$4aa zSvh--OmEK0vu4ZmrmUQGmP~KV%9*oddP7#8Ia8+BXXT7DWqO^GlVo~rR!*NG(`&MF z+H{#-ot3yAczRVteiAirdKNYkW8=0$_e-@%Cd4C%F?VH zJ3*#PlpH71R2{=B9PaWU{TRl`u?8sO=oti+|i)AO?uSNu*FW+g84ot~GKxW0F~AS-cM@ATZP z#MQjhbFvZ_@lMaq%I-a6`Ya{imFZbo*$vNOW>$9XCevqTWqwzgo{^PZ@@0B@R(8b0 zre$S^jxs$pE8BOG=_yKfkm<=;iA!y#CuLBb>1y}vId}$H@9cbDU#7F(**T79koC^a z$4z89>z$o8y8lu2?M7S?}!ZS|-z3@9bPsD$`l->|D4~ zrnBDJ*}fjq*48>hbD>FNB$XQ?Js4Tyfez0XkqfrC~MS{X>!dZYsxe^ zW|YzIWSZPEO0qC{Wt4!g2Ki)?Xl(MxDBYiBn*1?J$CYVv$0%)l3CJ0fblfz#Vh^R8 zCO^zQ8?S+!FmC?Ymg#C2%(09i)72iBV--WDvkn-G7&4u8!1hOYWmyM|B@CIacEB7f z7&4vpzgWPK>8$_7`h`qq{V$d;+%)-Lk6V_Tu6Dnkg$HH5FV-$(I_rJ0bOCAdzBDZn zD;JO^|4V>{3olI`7*}({x&@@k1=9vBTR@teFacIAAWd$V0E-roCPz$wH48|SD<;5_ z1^DEQv+y|hicxo!e1 zO@L3%n*b{l;FJ3%z`_LhaVz$d>? zfVBtk$@3Fn=>dH5{RCKf0H3@+0Tv#>C;v}?bqDY%0uW%?0ep%91Xy(dpP~Q(79GH+ zI6#0k2ki>U(JcY%-iJ!o`|6hqe5kH6> z0I!ShjBkr?jIY2u0H()BV-LVn_NbFm-Y$Y{T4 zmuSmqy(r{XzW>uJ-pAlMIE)qV8{i7q4qKrD#{g!)80Zh3ArER{{{Nl#3EmIzqIbl* z%iD*!|2A(UcGxTSW_jbip*Sm{gVzi*|B(Cqyk

;T7b=MlgRyVCZu)1n3J?m7h-8xpj(IKmg)*)$iHu~Sx@b%*~Lrz)k zf3lum{RQp+bTz+vE}yCo1-1R4wvU4k!kP4+RNEhRMEpg4K5*bse5&=3Vd4{`hl&$g z4;~@@to8rkyi%ucKk=bHwohO2f!4kIi1)Sb)m!{g>z=*Dds_GCDc;q(dk^uB(eH{s zXx**5cw6hP-Naj3=XVuvYTYGYyrFf+cf~QSJ9HGk*SdWd@w(9+#A{l&YcGy!-R52K zs@ARBiC2tnBVN|J)nM_G)-663FKV6FLcE}L^E~mq*3FuW=d^CxOgyV~lcwSstsA!z zPix(%v3SbpD)FS&^*~tXO7MefhOJX29?*MyUSAy4`gkL8zt$f&5%+2RZfkL`)<=ejd$fLN zthih2D;A5pwBEH$+^O{?rQ!~)FI*{Z*LwSUaX@=b8nqO+8Ny*pajV)7$1TMzS~qPj zZq^#D{yU?q#7$Z^{8Zeib^V57zt;7dh#QQqFZO9&`%7`X)-`L3y;|3(C-xX!Q(UKY J^gD6we*+%~Q?LL4 delta 297 zcmZo@U~O2yJVBn3cca2teMW=L@ASn2gxL8b82Eqkf8?LYU(TPypUfYzSx~@`ET&w=YPh3k^c+O6TA8KKn`bQ1G}G*nGMWi cVFgpHEMSU>8BDP=u>j55{J)-2VNrts0A;aV - + C:\Users\Devin\Repos\fmd_api\fmd_api - + - + @@ -18,41 +18,40 @@ - + - + - - + - + + - - - + + + - - - - - + + + + @@ -60,179 +59,168 @@ - - + + + - + - - + + - + - - - - + + + + - + - - + - + - + - + + - + - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + - - - - - - - - - - + + + + + + + + - - - - - - - - - - + + + + + + + + + + + - - - - - - + + + + + + + - - - - - + + + - - - - - - - - - + + + + + + + + + + - - - - + + + + - - + + - - - - - - - - - - - - + + - + + + + + + - - - - - - - - - - - - + + + + + + + + - - - - - - + + - - - - - - - - + + - - + + + + + + + + - + + + - - + + @@ -240,193 +228,216 @@ - - - - - - - - - + + + + + + + + - - + - + - + - - + + + + - - - + - + + + + - + + + - - - + + - - + - + + + + + + + + + + + + + + + + + + + + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - - - - + + + + + - + - - - - - - - - + + + + + + + + - + + + + + - + - - + + + + - - - + + - - - - + + - - + - - + - + + + + + - - - - - + + + + + + - - - - - + + + - + - - - - + + + + - - - - + + + + + + + + + + + + + + + + - + - - + - + - - - + + + @@ -434,90 +445,91 @@ - - - - + + + + - + - + - + - + - + - - - - - - + + + + + + - - - + + + - + - + - + - - - + + + - + - - - - + + + + + + + - - - + - - - - + + + - - - + + + + - + - - + - + + - + diff --git a/fmd_api/_version.py b/fmd_api/_version.py index 159d48b..0309ae2 100644 --- a/fmd_api/_version.py +++ b/fmd_api/_version.py @@ -1 +1 @@ -__version__ = "2.0.1" +__version__ = "2.0.2" diff --git a/tests/unit/test_coverage_improvements.py b/tests/unit/test_coverage_improvements.py new file mode 100644 index 0000000..911c265 --- /dev/null +++ b/tests/unit/test_coverage_improvements.py @@ -0,0 +1,1221 @@ +""" +Additional tests to improve code coverage to 95%+ +Focuses on uncovered branches and edge cases in client.py and device.py +""" +import json +import base64 +import pytest +from aioresponses import aioresponses +from cryptography.hazmat.primitives import serialization, hashes +from cryptography.hazmat.primitives.asymmetric import rsa, padding as asym_padding +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.backends import default_backend +from argon2.low_level import hash_secret_raw, Type +import aiohttp + +from fmd_api.client import FmdClient, CONTEXT_STRING_ASYM_KEY_WRAP +from fmd_api.device import Device +from fmd_api.exceptions import FmdApiException, OperationError + + +# ========================================== +# Test authentication helper methods +# ========================================== + +@pytest.mark.asyncio +async def test_hash_password_internal(): + """Test _hash_password generates correct format.""" + client = FmdClient("https://fmd.example.com") + result = client._hash_password("testpass", "dGVzdHNhbHQxMjM0NTY3OA") + + assert result.startswith("$argon2id$v=19$m=131072,t=1,p=4$") + assert "$" in result + parts = result.split("$") + assert len(parts) == 6 # empty, argon2id, v=19, params, salt, hash + + +@pytest.mark.asyncio +async def test_load_private_key_from_pem(): + """Test _load_private_key_from_bytes with PEM format.""" + client = FmdClient("https://fmd.example.com") + + # Generate a test RSA key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend() + ) + + pem_bytes = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption() + ) + + loaded_key = client._load_private_key_from_bytes(pem_bytes) + assert loaded_key is not None + + +@pytest.mark.asyncio +async def test_load_private_key_from_der(): + """Test _load_private_key_from_bytes with DER format (fallback path).""" + client = FmdClient("https://fmd.example.com") + + # Generate a test RSA key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend() + ) + + der_bytes = private_key.private_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption() + ) + + loaded_key = client._load_private_key_from_bytes(der_bytes) + assert loaded_key is not None + + +# ========================================== +# Test JSON parsing fallback paths +# ========================================== + +@pytest.mark.asyncio +async def test_json_parse_error_fallback_to_text(): + """Test that JSONDecodeError triggers fallback to text response.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + # Return text that will trigger JSONDecodeError + m.put( + "https://fmd.example.com/api/v1/salt", + body='"invalid json', # Missing closing quote + content_type="application/json" + ) + + try: + # Should fall back to text and return the raw string + result = await client._make_api_request("PUT", "/api/v1/salt", {"IDT": "test", "Data": ""}) + assert result == '"invalid json' + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_json_missing_data_key_fallback(): + """Test KeyError when JSON response lacks 'Data' key.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + # Return JSON without 'Data' key + m.put( + "https://fmd.example.com/api/v1/salt", + payload={"error": "something"}, + content_type="application/json" + ) + + try: + # Should catch KeyError and fall back to text + result = await client._make_api_request("PUT", "/api/v1/salt", {"IDT": "test", "Data": ""}) + # aioresponses returns the payload dict as-is when no 'Data' + assert isinstance(result, (str, dict)) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_empty_text_response_warning(): + """Test that empty text response triggers warning log.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + # Return JSON with Data key but with empty string value + m.put( + "https://fmd.example.com/api/v1/salt", + payload={"Data": ""}, + content_type="application/json" + ) + + try: + result = await client._make_api_request("PUT", "/api/v1/salt", {"IDT": "test", "Data": ""}) + # Empty response should return empty string + assert result == "" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_expect_json_false_path(): + """Test expect_json=False returns text directly.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + m.post( + "https://fmd.example.com/api/v1/command", + body="Command received", + status=200 + ) + + try: + result = await client._make_api_request( + "POST", + "/api/v1/command", + {"IDT": "token", "Data": "test"}, + expect_json=False + ) + assert result == "Command received" + finally: + await client.close() + + +# ========================================== +# Test connection error retry logic +# ========================================== + +@pytest.mark.asyncio +async def test_connection_error_retry_with_backoff(monkeypatch): + """Test ClientConnectionError triggers retry with backoff.""" + client = FmdClient("https://fmd.example.com", max_retries=2, backoff_base=0.1, jitter=False) + client.access_token = "token" + + slept = [] + async def fake_sleep(seconds): + slept.append(seconds) + + monkeypatch.setattr("asyncio.sleep", fake_sleep) + + await client._ensure_session() + + with aioresponses() as m: + # First two attempts: connection error, third: success + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + exception=aiohttp.ClientConnectionError("Connection failed") + ) + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + exception=aiohttp.ClientConnectionError("Connection failed") + ) + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + payload={"Data": "0"} + ) + + try: + result = await client.get_locations() + assert result == [] + # Should have two backoff sleeps: 0.1, 0.2 + assert len(slept) == 2 + assert slept[0] == 0.1 + assert slept[1] == 0.2 + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_connection_error_exhausted_retries(monkeypatch): + """Test connection error raises FmdApiException when retries exhausted.""" + client = FmdClient("https://fmd.example.com", max_retries=1, backoff_base=0.1, jitter=False) + client.access_token = "token" + + slept = [] + async def fake_sleep(seconds): + slept.append(seconds) + + monkeypatch.setattr("asyncio.sleep", fake_sleep) + + await client._ensure_session() + + with aioresponses() as m: + # All attempts fail + for _ in range(3): + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + exception=aiohttp.ClientConnectionError("Connection failed") + ) + + try: + with pytest.raises(FmdApiException, match="API request failed"): + await client.get_locations() + # Should have 1 retry sleep + assert len(slept) == 1 + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_connection_error_no_retry_for_unsafe_command(): + """Test connection error doesn't retry for unsafe command POST.""" + client = FmdClient("https://fmd.example.com", max_retries=3) + client.access_token = "token" + + # Set up private key for send_command + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend() + ) + client.private_key = private_key + + await client._ensure_session() + + with aioresponses() as m: + # Connection error on command endpoint + m.post( + "https://fmd.example.com/api/v1/command", + exception=aiohttp.ClientConnectionError("Connection failed") + ) + + try: + with pytest.raises(FmdApiException, match="Failed to send command"): + await client.send_command("ring") + finally: + await client.close() + + +# ========================================== +# Test export_data_zip edge cases +# ========================================== + +@pytest.mark.asyncio +async def test_export_zip_png_detection(): + """Test PNG magic byte detection in export_data_zip.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + + # Set up private key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=3072, + backend=default_backend() + ) + client.private_key = private_key + + # Create PNG image bytes (PNG magic bytes + minimal data) + png_data = b'\x89PNG\r\n\x1a\n' + b'\x00' * 100 + png_b64 = base64.b64encode(png_data).decode('utf-8') + + # Double-encode as per FMD picture format + session_key = b'\x00' * 32 + aesgcm = AESGCM(session_key) + iv = b'\x01' * 12 + + # Encrypt the base64 string + ciphertext = aesgcm.encrypt(iv, png_b64.encode('utf-8'), None) + + # Build encrypted blob + public_key = private_key.public_key() + session_key_packet = public_key.encrypt( + session_key, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None + ) + ) + + blob = session_key_packet + iv + ciphertext + blob_b64 = base64.b64encode(blob).decode('utf-8') + + await client._ensure_session() + + with aioresponses() as m: + # No locations + m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "0"}) + # One PNG picture + m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": [blob_b64]}) + + try: + import tempfile + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: + output_path = tmp.name + + result = await client.export_data_zip(output_path, include_pictures=True) + assert result == output_path + + # Verify ZIP contains PNG + import zipfile + with zipfile.ZipFile(output_path, 'r') as zf: + files = zf.namelist() + assert 'pictures/manifest.json' in files + assert any('picture_' in f and f.endswith('.png') for f in files) + + import os + os.unlink(output_path) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_export_zip_picture_decrypt_error(): + """Test export handles picture decryption errors gracefully.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + + # Set up private key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend() + ) + client.private_key = private_key + + await client._ensure_session() + + with aioresponses() as m: + m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "0"}) + # Invalid picture blob (too small) + m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": ["invalid"]}) + + try: + import tempfile + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: + output_path = tmp.name + + result = await client.export_data_zip(output_path, include_pictures=True) + + # Should complete despite error + import zipfile + with zipfile.ZipFile(output_path, 'r') as zf: + manifest = json.loads(zf.read('pictures/manifest.json')) + # Error should be recorded + assert 'error' in manifest[0] + + import os + os.unlink(output_path) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_export_zip_location_decrypt_error(): + """Test export handles location decryption errors gracefully.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + + # Set up private key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend() + ) + client.private_key = private_key + + await client._ensure_session() + + with aioresponses() as m: + # One invalid location + m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "1"}) + m.put("https://fmd.example.com/api/v1/location", payload={"Data": "tooshort"}) + # No pictures + m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": []}) + + try: + import tempfile + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: + output_path = tmp.name + + result = await client.export_data_zip(output_path, include_pictures=False) + + # Should complete with error recorded + import zipfile + with zipfile.ZipFile(output_path, 'r') as zf: + locations = json.loads(zf.read('locations.json')) + assert 'error' in locations[0] + assert locations[0]['index'] == 0 + + import os + os.unlink(output_path) + finally: + await client.close() + + +# ========================================== +# Test device.py missing lines +# ========================================== + +@pytest.mark.asyncio +async def test_device_download_photo_decode_error(): + """Test Device.download_photo handles decode errors (line 137-138).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + + # Set up private key with 3072-bit key to get 384-byte RSA packet + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=3072, + backend=default_backend() + ) + client.private_key = private_key + + device = Device(client, "test_device") + + # Create an invalid blob (will decrypt but not be valid base64) + session_key = b'\x00' * 32 + aesgcm = AESGCM(session_key) + iv = b'\x01' * 12 + + # Invalid inner data (not valid base64) + ciphertext = aesgcm.encrypt(iv, b'not-base64-data!!!', None) + + public_key = private_key.public_key() + session_key_packet = public_key.encrypt( + session_key, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None + ) + ) + + blob = session_key_packet + iv + ciphertext + blob_b64 = base64.b64encode(blob).decode('utf-8') + + with pytest.raises(OperationError, match="Failed to decode picture blob"): + await device.download_photo(blob_b64) + + +@pytest.mark.asyncio +async def test_device_get_history_decrypt_error(): + """Test Device.get_history handles decrypt errors (line 99-101).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + + # Set up private key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend() + ) + client.private_key = private_key + + device = Device(client, "test_device") + + await client._ensure_session() + + with aioresponses() as m: + m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "1"}) + m.put("https://fmd.example.com/api/v1/location", payload={"Data": "invalid"}) + + try: + with pytest.raises(OperationError, match="Failed to decrypt/parse location blob"): + async for loc in device.get_history(limit=1): + pass + finally: + await client.close() + + + + + +# ========================================== +# Test helper functions indirectly through client behavior +# ========================================== + +@pytest.mark.asyncio +async def test_retry_after_header_parsing_indirectly(): + """Test Retry-After header parsing through actual 429 response.""" + client = FmdClient("https://fmd.example.com", max_retries=2) + client.access_token = "token" + + await client._ensure_session() + + with aioresponses() as m: + # Test with valid Retry-After number + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + status=429, + headers={"Retry-After": "5"} + ) + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + payload={"Data": "0"} + ) + + try: + await client.get_locations() + # If it succeeds, Retry-After was parsed correctly + finally: + await client.close() + + +# ========================================== +# Additional edge cases +# ========================================== + +@pytest.mark.asyncio +async def test_decrypt_blob_with_missing_private_key(): + """Test decrypt_data_blob raises when private_key is None.""" + client = FmdClient("https://fmd.example.com") + # Don't set private_key + + # Use a valid base64 string that's long enough + dummy_blob = base64.b64encode(b'\x00' * 400).decode('utf-8') + with pytest.raises(FmdApiException, match="Private key not loaded"): + client.decrypt_data_blob(dummy_blob) + + +@pytest.mark.asyncio +async def test_send_command_with_missing_private_key(): + """Test send_command raises when private_key is None.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + try: + with pytest.raises(FmdApiException, match="Private key not loaded"): + await client.send_command("ring") + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_client_error_generic(): + """Test generic ClientError handling.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + exception=aiohttp.ClientError("Generic client error") + ) + + try: + with pytest.raises(FmdApiException, match="API request failed"): + await client.get_locations() + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_value_error_in_response_parsing(): + """Test ValueError in response parsing.""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + # Return JSON that will cause ValueError when parsing int + m.put( + "https://fmd.example.com/api/v1/locationDataSize", + payload={"Data": "not-a-number"} + ) + + try: + with pytest.raises(Exception): # int() will raise ValueError, caught and re-raised + await client.get_locations() + finally: + await client.close() + + +# ========================================== +# Additional tests to reach 95% coverage +# ========================================== + +@pytest.mark.asyncio +async def test_authenticate_full_flow(): + """Test complete authenticate flow including internal methods (lines 163-211).""" + client = FmdClient("https://fmd.example.com") + + with aioresponses() as m: + # Mock salt retrieval + m.put("https://fmd.example.com/api/v1/salt", payload={"Data": base64.b64encode(b'\x00' * 16).decode()}) + # Mock token request + m.put("https://fmd.example.com/api/v1/requestAccess", payload={"Data": "test_token"}) + # Mock private key retrieval + # Create a simple encrypted key blob + password = "testpass" + salt = b'\x00' * 16 + iv = b'\x01' * 12 + + # Create a dummy private key + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend()) + privkey_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption() + ) + + # Encrypt the private key + password_bytes = (CONTEXT_STRING_ASYM_KEY_WRAP + password).encode("utf-8") + aes_key = hash_secret_raw( + secret=password_bytes, salt=salt, time_cost=1, memory_cost=131072, parallelism=4, hash_len=32, type=Type.ID + ) + aesgcm = AESGCM(aes_key) + ciphertext = aesgcm.encrypt(iv, privkey_pem, None) + encrypted_blob = salt + iv + ciphertext + + m.put("https://fmd.example.com/api/v1/key", payload={"Data": base64.b64encode(encrypted_blob).decode()}) + + try: + await client.authenticate("testid", password, 3600) + assert client.access_token == "test_token" + assert client.private_key is not None + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_429_with_retry_after_header(): + """Test 429 response with Retry-After header (lines 305-312).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 1 + await client._ensure_session() + + with aioresponses() as m: + # First request returns 429 with Retry-After + m.get( + "https://fmd.example.com/api/v1/test", + status=429, + headers={"Retry-After": "1"} + ) + # Second request succeeds + m.get( + "https://fmd.example.com/api/v1/test", + payload={"Data": "success"} + ) + + try: + result = await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + assert result == "success" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_500_error_retry(): + """Test 500 server error triggers retry (lines 353-358).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 1 + await client._ensure_session() + + with aioresponses() as m: + # First request returns 500 + m.get( + "https://fmd.example.com/api/v1/test", + status=500 + ) + # Second request succeeds + m.get( + "https://fmd.example.com/api/v1/test", + payload={"Data": "success"} + ) + + try: + result = await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + assert result == "success" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_negative_retry_after_header(): + """Test Retry-After with negative value (lines 703-707).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 1 + await client._ensure_session() + + with aioresponses() as m: + # First request returns 429 with invalid negative Retry-After + m.get( + "https://fmd.example.com/api/v1/test", + status=429, + headers={"Retry-After": "-5"} + ) + # Second request succeeds + m.get( + "https://fmd.example.com/api/v1/test", + payload={"Data": "success"} + ) + + try: + result = await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + assert result == "success" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_http_date_retry_after(): + """Test Retry-After with HTTP-date format (lines 709-711).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 1 + await client._ensure_session() + + with aioresponses() as m: + # First request returns 429 with HTTP-date Retry-After + m.get( + "https://fmd.example.com/api/v1/test", + status=429, + headers={"Retry-After": "Wed, 21 Oct 2025 07:28:00 GMT"} + ) + # Second request succeeds + m.get( + "https://fmd.example.com/api/v1/test", + payload={"Data": "success"} + ) + + try: + result = await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + assert result == "success" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_backoff_without_jitter(): + """Test backoff calculation without jitter (lines 715-718).""" + client = FmdClient("https://fmd.example.com", jitter=False) + client.access_token = "token" + client.max_retries = 2 + await client._ensure_session() + + with aioresponses() as m: + # First request returns 500 + m.get("https://fmd.example.com/api/v1/test", status=500) + # Second request returns 500 + m.get("https://fmd.example.com/api/v1/test", status=500) + # Third request succeeds + m.get("https://fmd.example.com/api/v1/test", payload={"Data": "success"}) + + try: + result = await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + assert result == "success" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_device_internal_parse_location_error(): + """Test that _parse_location_blob raises RuntimeError (device.py line 23).""" + from fmd_api.device import _parse_location_blob + + with pytest.raises(RuntimeError, match="should not be called directly"): + _parse_location_blob("dummy_blob") + + +@pytest.mark.asyncio +async def test_get_pictures_with_specific_count(): + """Test get_pictures with specific num_to_get (lines 477-478).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + # Set up private key for decryption + private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend()) + client.private_key = private_key + + with aioresponses() as m: + # Mock response with 10 pictures + pictures_list = [f"picture{i}" for i in range(10)] + m.put( + "https://fmd.example.com/api/v1/pictureDataSize", + payload={"Data": "10"} + ) + m.put( + "https://fmd.example.com/api/v1/pictures", + payload={"Data": pictures_list} + ) + + try: + # Request only 3 pictures + result = await client.get_pictures(num_to_get=3) + assert len(result) == 3 + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_exhausted_retries_on_500(): + """Test exhausting retries on repeated 500 errors (lines 353-358).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 2 + await client._ensure_session() + + with aioresponses() as m: + # All requests return 500 + for _ in range(5): + m.get("https://fmd.example.com/api/v1/test", status=500) + + try: + with pytest.raises(FmdApiException, match="API request failed"): + await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_compute_backoff_with_jitter(): + """Test _compute_backoff with jitter enabled (line 715-718).""" + from fmd_api.client import _compute_backoff + + # With jitter, result should be between 0 and calculated delay + for attempt in range(3): + delay = _compute_backoff(1.0, attempt, 10.0, True) + expected_max = min(10.0, 1.0 * (2 ** attempt)) + assert 0 <= delay <= expected_max + + +@pytest.mark.asyncio +async def test_429_exhausted_retries(): + """Test 429 with exhausted retries (lines 305-306).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 0 # No retries + await client._ensure_session() + + with aioresponses() as m: + m.get( + "https://fmd.example.com/api/v1/test", + status=429 + ) + + try: + with pytest.raises(FmdApiException, match="Rate limited.*retries exhausted"): + await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_streaming_response(): + """Test streaming response path (line 374-376).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + m.get( + "https://fmd.example.com/api/v1/test", + body="streaming content", + content_type="text/plain" + ) + + try: + result = await client._make_api_request( + "GET", "/api/v1/test", + {"IDT": "test", "Data": ""}, + stream=True # Request streaming response + ) + # Should return the response object itself + assert result is not None + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_get_pictures_all_count(): + """Test get_pictures with num_to_get=-1 (all pictures) (lines 477-478).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + # Set up private key + private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend()) + client.private_key = private_key + + with aioresponses() as m: + # Mock response with 5 pictures + pictures_list = [f"picture{i}" for i in range(5)] + m.put( + "https://fmd.example.com/api/v1/pictureDataSize", + payload={"Data": "5"} + ) + m.put( + "https://fmd.example.com/api/v1/pictures", + payload={"Data": pictures_list} + ) + + try: + # Request all pictures (num_to_get=-1) + result = await client.get_pictures(num_to_get=-1) + assert len(result) == 5 + finally: + await client.close() + + +# ========================================== +# Final push to 100% coverage +# ========================================== + +@pytest.mark.asyncio +async def test_500_error_exhausted_retries_raises(): + """Test 500 error with exhausted retries (lines 353-358).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 1 + await client._ensure_session() + + with aioresponses() as m: + # All requests return 500 + for _ in range(3): + m.post("https://fmd.example.com/api/v1/test", status=500) + + try: + with pytest.raises(FmdApiException, match="API request failed"): + await client._make_api_request("POST", "/api/v1/test", {"IDT": "test", "Data": ""}) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_expect_json_false_returns_text(): + """Test expect_json=False returns text directly (line 368).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + m.post( + "https://fmd.example.com/api/v1/command", + body="Command executed", + content_type="text/plain" + ) + + try: + result = await client._make_api_request( + "POST", "/api/v1/command", + {"IDT": "test", "Data": ""}, + expect_json=False + ) + assert result == "Command executed" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_response_parsing_key_error(): + """Test KeyError/ValueError in response parsing (lines 392-393).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + # Return invalid response that will cause parsing error outside the JSON block + m.put( + "https://fmd.example.com/api/v1/test", + payload={"Data": {"nested": "value"}}, # Valid JSON but might cause issues downstream + content_type="application/json" + ) + + try: + # This should work without errors + result = await client._make_api_request("PUT", "/api/v1/test", {"IDT": "test", "Data": ""}) + assert result == {"nested": "value"} + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_get_pictures_returns_list_when_all(): + """Test get_pictures returns full list when num_to_get=-1 (line 480).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + # Set up private key + private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend()) + client.private_key = private_key + + with aioresponses() as m: + # Mock response with 10 pictures + pictures_list = [f"picture{i}" for i in range(10)] + m.put( + "https://fmd.example.com/api/v1/pictureDataSize", + payload={"Data": "10"} + ) + m.put( + "https://fmd.example.com/api/v1/pictures", + payload={"Data": pictures_list} + ) + + try: + # Request all pictures explicitly + result = await client.get_pictures(num_to_get=-1) + # Should return all 10 pictures + assert len(result) == 10 + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_export_zip_default_jpg_extension(): + """Test export_data_zip defaults to jpg for unknown image types (line 567).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + # Set up private key for decryption + private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend()) + client.private_key = private_key + + # Create an encrypted blob with unknown image format (not PNG or JPEG) + session_key = b'\x00' * 32 + aesgcm = AESGCM(session_key) + iv = b'\x01' * 12 + + # Create image data that doesn't match PNG or JPEG magic bytes + unknown_image = b'\x00\x00\x00\x00UNKNOWN' + b'\x00' * 20 + image_b64 = base64.b64encode(unknown_image).decode('utf-8') + + # Encrypt it + ciphertext = aesgcm.encrypt(iv, image_b64.encode('utf-8'), None) + + public_key = private_key.public_key() + session_key_packet = public_key.encrypt( + session_key, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None + ) + ) + + blob = session_key_packet + iv + ciphertext + blob_b64 = base64.b64encode(blob).decode('utf-8') + + with aioresponses() as m: + m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "0"}) + m.put("https://fmd.example.com/api/v1/pictureDataSize", payload={"Data": "1"}) + m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": [blob_b64]}) + + try: + import tempfile + import zipfile + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: + output_path = tmp.name + + await client.export_data_zip(output_path, include_pictures=True) + + # Verify the file was created and contains jpg file + with zipfile.ZipFile(output_path, 'r') as zf: + names = zf.namelist() + # Should have defaulted to .jpg extension + assert any('.jpg' in name for name in names) + + # Cleanup + import os + if os.path.exists(output_path): + os.unlink(output_path) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_mask_token_with_none(): + """Test _mask_token with None input (line 685-686).""" + from fmd_api.client import _mask_token + + result = _mask_token(None) + assert result == "" + + +@pytest.mark.asyncio +async def test_mask_token_with_short(): + """Test _mask_token with short token (line 687-688).""" + from fmd_api.client import _mask_token + + result = _mask_token("ab", show_chars=5) + assert result == "***" + + +@pytest.mark.asyncio +async def test_mask_token_with_long(): + """Test _mask_token with long token (line 689).""" + from fmd_api.client import _mask_token + + result = _mask_token("verylongtoken123456", show_chars=4) + assert result.startswith("very") + assert result.endswith("...***") + + +@pytest.mark.asyncio +async def test_parse_retry_after_with_invalid(): + """Test _parse_retry_after with invalid input (line 703).""" + from fmd_api.client import _parse_retry_after + + # None input + result = _parse_retry_after(None) + assert result is None + + # Invalid string + result = _parse_retry_after("invalid") + assert result is None + + +@pytest.mark.asyncio +async def test_compute_backoff_with_jitter_randomness(): + """Test _compute_backoff with jitter produces values in range (line 717-718).""" + from fmd_api.client import _compute_backoff + + # With jitter=True, should return random value between 0 and delay + delays = [_compute_backoff(1.0, 0, 10.0, True) for _ in range(10)] + + # All should be >= 0 and <= 1.0 (base * 2^0) + assert all(0 <= d <= 1.0 for d in delays) + + # With enough samples, should have some variation (not all the same) + # (This might fail in rare cases but is statistically very unlikely) + assert len(set(delays)) > 1 or delays[0] == 0 # Allow all zeros as edge case + + +@pytest.mark.asyncio +async def test_502_error_with_exhausted_retries(): + """Test 502 error exhausts retries and continues to raise (lines 353-358).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + client.max_retries = 1 + await client._ensure_session() + + with aioresponses() as m: + # All requests return 502 + for _ in range(5): + m.get("https://fmd.example.com/api/v1/test", status=502) + + try: + with pytest.raises(FmdApiException, match="API request failed"): + await client._make_api_request("GET", "/api/v1/test", {"IDT": "test", "Data": ""}) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_non_json_response_with_expect_json_false(): + """Test non-JSON response when expect_json=False (line 368).""" + client = FmdClient("https://fmd.example.com") + client.access_token = "token" + await client._ensure_session() + + with aioresponses() as m: + m.put( + "https://fmd.example.com/api/v1/test", + body="plain text response", + content_type="text/plain" + ) + + try: + result = await client._make_api_request( + "PUT", "/api/v1/test", + {"IDT": "test", "Data": ""}, + expect_json=False + ) + assert result == "plain text response" + finally: + await client.close()